diff --git a/cueapi/cli.py b/cueapi/cli.py
index 8b998be..5735001 100644
--- a/cueapi/cli.py
+++ b/cueapi/cli.py
@@ -109,6 +109,27 @@ def quickstart(ctx: click.Context) -> None:
         "Strictly 1:1 chaining; the target cue is validated at create time."
     ),
 )
+@click.option(
+    "--require-payload-override/--no-require-payload-override",
+    "require_payload_override",
+    default=None,
+    help=(
+        "Require payload_override on every fire (server-side enforcement, hosted PR #590). "
+        "Use on team-comm cues where the payload IS the message; leave unset for cron-style "
+        "cues that rely on the stored cue.payload. Fires without payload_override are rejected "
+        "with HTTP 400 payload_override_required."
+    ),
+)
+@click.option(
+    "--required-keys",
+    "required_keys",
+    default=None,
+    help=(
+        "Comma-separated keys that must be present in the resolved override on fire (post-merge). "
+        "Missing keys yield HTTP 400 missing_required_payload_keys. Implies --require-payload-override "
+        "in spirit but doesn't set it; pass both for full enforcement. Empty string sends an empty list."
+    ),
+)
 @click.pass_context
 def create(
     ctx: click.Context,
@@ -127,6 +148,8 @@ def create(
     catch_up: Optional[str],
     verification: Optional[str],
     on_success_fire: Optional[str],
+    require_payload_override: Optional[bool],
+    required_keys: Optional[str],
 ) -> None:
     """Create a new cue."""
     if cron and at_time:
@@ -193,6 +216,18 @@ def create(
     if on_success_fire:
         body["on_success_fire"] = on_success_fire
 
+    # Hosted PR #590: per-cue opt-in enforcement of payload_override on /fire.
+    # `require_payload_override=None` means "not specified" — omit from body so
+    # the server's default (false) applies on create.
+    if require_payload_override is not None:
+        body["require_payload_override"] = require_payload_override
+
+    # `required_keys=None` → omit. Empty string → send `[]` (explicit clear).
+    # Non-empty string → split, trim, drop empties.
+    if required_keys is not None:
+        parsed_keys = [k.strip() for k in required_keys.split(",") if k.strip()]
+        body["required_payload_keys"] = parsed_keys
+
     try:
         with CueAPIClient(api_key=ctx.obj.get("api_key"), profile=ctx.obj.get("profile")) as client:
             resp = client.post("/cues", json=body)
@@ -491,6 +526,25 @@ def fire(ctx: click.Context, cue_id: str, payload_override: Optional[str], merge
     default=False,
     help="Clear on_success_fire (disable chaining). Mutually exclusive with --on-success-fire.",
 )
+@click.option(
+    "--require-payload-override/--no-require-payload-override",
+    "require_payload_override",
+    default=None,
+    help=(
+        "Toggle server-side enforcement of payload_override on fire (hosted PR #590). "
+        "--require-payload-override turns it on; --no-require-payload-override turns it off. "
+        "Omit to leave unchanged."
+    ),
+)
+@click.option(
+    "--required-keys",
+    "required_keys",
+    default=None,
+    help=(
+        "Comma-separated keys that must be present in the resolved override on fire. "
+        "Empty string sends `[]` (explicit clear). Omit to leave unchanged."
+    ),
+)
 @click.pass_context
 def update(ctx: click.Context, cue_id: str, name: Optional[str], cron: Optional[str],
            url: Optional[str], payload: Optional[str], description: Optional[str],
@@ -501,7 +555,9 @@ def update(ctx: click.Context, cue_id: str, name: Optional[str], cron: Optional[
            catch_up: Optional[str],
            verification: Optional[str],
            on_success_fire: Optional[str],
-           clear_on_success_fire: bool) -> None:
+           clear_on_success_fire: bool,
+           require_payload_override: Optional[bool],
+           required_keys: Optional[str]) -> None:
     """Update an existing cue."""
     if on_success_fire and clear_on_success_fire:
         raise click.UsageError("--on-success-fire and --clear-on-success-fire are mutually exclusive.")
@@ -550,6 +606,15 @@ def update(ctx: click.Context, cue_id: str, name: Optional[str], cron: Optional[
         # null in JSON.
         body["on_success_fire"] = None
 
+    # Hosted PR #590: tri-state on update — None omits, True/False sends.
+    if require_payload_override is not None:
+        body["require_payload_override"] = require_payload_override
+
+    # required_keys: None omits; empty string sends []; non-empty splits.
+    if required_keys is not None:
+        parsed_keys = [k.strip() for k in required_keys.split(",") if k.strip()]
+        body["required_payload_keys"] = parsed_keys
+
     if not body:
         raise click.UsageError("Must specify at least one field to update.")
 
@@ -791,6 +856,13 @@ def executions_get(ctx: click.Context, execution_id: str) -> None:
                 echo_info("HTTP status:", str(ex["http_status"]))
             if ex.get("error_message"):
                 echo_info("Error:", ex["error_message"])
+            # Effective payload (hosted PR #589): the JSON the handler /
+            # webhook actually saw at delivery time. Falls back to the
+            # parent cue's stored payload when no per-fire override was
+            # set. Surfaced for forensics — what was delivered, not what
+            # the cue's stored default looks like at query time.
+            if ex.get("payload") is not None:
+                echo_info("Payload:", json.dumps(ex["payload"], indent=2, sort_keys=True))
             click.echo()
     except click.ClickException as e:
         click.echo(str(e))
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 308e209..8edf325 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -1142,3 +1142,270 @@ def test_update_catch_up_validated_by_click():
         or "invalid" in result.output.lower()
         or "run_once_if_missed" in result.output
     )
+
+
+# --- payload_override enforcement flags (hosted PR #590) ---
+
+
+def test_create_require_payload_override_in_help():
+    result = runner.invoke(main, ["create", "--help"])
+    assert result.exit_code == 0
+    assert "--require-payload-override" in result.output
+    assert "--no-require-payload-override" in result.output
+
+
+def test_create_required_keys_in_help():
+    result = runner.invoke(main, ["create", "--help"])
+    assert result.exit_code == 0
+    assert "--required-keys" in result.output
+
+
+def test_update_require_payload_override_in_help():
+    result = runner.invoke(main, ["update", "--help"])
+    assert result.exit_code == 0
+    assert "--require-payload-override" in result.output
+    assert "--no-require-payload-override" in result.output
+
+
+def test_update_required_keys_in_help():
+    result = runner.invoke(main, ["update", "--help"])
+    assert result.exit_code == 0
+    assert "--required-keys" in result.output
+
+
+# --- body-construction unit tests for new flags ---
+#
+# These mock the HTTP layer (CueAPIClient.post / .patch) and assert that the
+# CLI body matches the hosted-API spec. Cheap insurance against flag-wiring
+# regressions when refactoring create/update body builders.
+
+
+class _FakeResp:
+    def __init__(self, status_code: int, payload: dict):
+        self.status_code = status_code
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+
+class _FakeClient:
+    def __init__(self):
+        self.posted: list = []
+        self.patched: list = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        pass
+
+    def post(self, path, json=None, **_):
+        self.posted.append((path, json))
+        return _FakeResp(201, {"id": "cue_test", "status": "active", "next_run": None})
+
+    def patch(self, path, json=None, **_):
+        self.patched.append((path, json))
+        return _FakeResp(200, {"id": "cue_test", "name": json.get("name", "x") if json else "x"})
+
+    def get(self, *_, **__):
+        # Not used by these tests but defined so the context-manager surface
+        # matches CueAPIClient.
+        return _FakeResp(200, {})
+
+
+def _patched_client(monkeypatch, client_holder):
+    """Patch CueAPIClient in cueapi.cli to return a captured FakeClient."""
+    import cueapi.cli as cli_mod
+
+    def fake_factory(*_, **__):
+        client_holder["client"] = _FakeClient()
+        return client_holder["client"]
+
+    monkeypatch.setattr(cli_mod, "CueAPIClient", fake_factory)
+
+
+def test_create_require_payload_override_true_sends_field(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        [
+            "create",
+            "--name", "team-comm",
+            "--cron", "0 9 * * *",
+            "--worker",
+            "--require-payload-override",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].posted[-1][1]
+    assert body.get("require_payload_override") is True
+
+
+def test_create_no_require_payload_override_sends_false(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        [
+            "create",
+            "--name", "cron-cue",
+            "--cron", "0 9 * * *",
+            "--worker",
+            "--no-require-payload-override",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].posted[-1][1]
+    assert body.get("require_payload_override") is False
+
+
+def test_create_omits_require_payload_override_when_unset(monkeypatch):
+    # Default None must NOT appear in the body — server-side default is the
+    # source of truth for "did the caller specify this?" Pinning this
+    # behavior so a refactor can't silently start sending false.
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        ["create", "--name", "x", "--cron", "0 9 * * *", "--worker"],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].posted[-1][1]
+    assert "require_payload_override" not in body
+
+
+def test_create_required_keys_splits_and_trims(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        [
+            "create",
+            "--name", "team-comm",
+            "--cron", "0 9 * * *",
+            "--worker",
+            "--required-keys", " task ,message,  token",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].posted[-1][1]
+    assert body.get("required_payload_keys") == ["task", "message", "token"]
+
+
+def test_create_required_keys_empty_string_sends_empty_list(monkeypatch):
+    # Empty string is the "explicit clear" path. Pin so a future refactor
+    # doesn't drop the body field entirely (which would mean "leave
+    # unchanged" not "clear").
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        [
+            "create",
+            "--name", "x",
+            "--cron", "0 9 * * *",
+            "--worker",
+            "--required-keys", "",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].posted[-1][1]
+    assert body.get("required_payload_keys") == []
+
+
+def test_update_require_payload_override_tri_state(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        ["update", "cue_test", "--require-payload-override"],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].patched[-1][1]
+    assert body.get("require_payload_override") is True
+
+
+def test_update_no_require_payload_override_sends_false(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        ["update", "cue_test", "--no-require-payload-override"],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].patched[-1][1]
+    assert body.get("require_payload_override") is False
+
+
+def test_update_required_keys_works(monkeypatch):
+    holder: dict = {}
+    _patched_client(monkeypatch, holder)
+    result = runner.invoke(
+        main,
+        ["update", "cue_test", "--required-keys", "a,b"],
+    )
+    assert result.exit_code == 0, result.output
+    body = holder["client"].patched[-1][1]
+    assert body.get("required_payload_keys") == ["a", "b"]
+
+
+# --- effective payload display on `executions get` (hosted PR #589) ---
+
+
+def test_executions_get_displays_payload_when_present(monkeypatch):
+    import cueapi.cli as cli_mod
+
+    fake_response = {
+        "id": "exec_abc",
+        "cue_id": "cue_xyz",
+        "status": "success",
+        "scheduled_for": "2026-05-04T10:00:00Z",
+        "attempts": 1,
+        "payload": {"task": "demo", "message": "hello"},
+    }
+
+    class _GetClient:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_):
+            pass
+
+        def get(self, path, **_):
+            return _FakeResp(200, fake_response)
+
+    monkeypatch.setattr(cli_mod, "CueAPIClient", lambda *_, **__: _GetClient())
+    result = runner.invoke(main, ["executions", "get", "exec_abc"])
+    assert result.exit_code == 0, result.output
+    assert "Payload:" in result.output
+    # Pretty-print with indent=2, sort_keys=True — pin the keys are visible.
+    assert "task" in result.output
+    assert "demo" in result.output
+
+
+def test_executions_get_omits_payload_when_null(monkeypatch):
+    import cueapi.cli as cli_mod
+
+    fake_response = {
+        "id": "exec_abc",
+        "cue_id": "cue_xyz",
+        "status": "pending",
+        "payload": None,
+    }
+
+    class _GetClient:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_):
+            pass
+
+        def get(self, path, **_):
+            return _FakeResp(200, fake_response)
+
+    monkeypatch.setattr(cli_mod, "CueAPIClient", lambda *_, **__: _GetClient())
+    result = runner.invoke(main, ["executions", "get", "exec_abc"])
+    assert result.exit_code == 0, result.output
+    assert "Payload:" not in result.output