audit-D16: operator key must never be an agent identity

Background: CAYCE_OPS_OPERATOR_PUBKEY was set to Cayce nostr pubkey —
Cayce was approving itself. Signature was technically valid but semantically
null: human approval signed by the agent is not human approval.

- signing.py: validate_no_self_approval() rejects the operator pubkey when
  it matches (a) the pubkey derived from any private key in the config,
  or (b) any pubkey in CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS deny-list.
- executor/maild/publisher _signing() now calls the validation, so every
  daemon and every CLI approve-* path refuses to operate (D16).
- tests: 6 new (agent-key refusal, human key passes, empty passes,
  explicit denylist, end-to-end daemon refusal, derivation roundtrip).
- docs/DECISIONS.md: D16.

The operator signature is now produced OUTSIDE the VPS. CAYCE_OPS_OPERATOR_PUBKEY
must stay empty until the external signer exists — daemons refuse, which is
correct. The Cayce nsec itself is NOT touched (still the agent identity for
publishing and anchoring).

README.md

@@ -1,8 +1,10 @@
-# Cayce is an agent-operated project run by Gustavo.
+# Cayce Ops
 
-Cayce Ops is the autonomous operations stack behind Cayce — an agent that runs
+Cayce Ops is the autonomous operations stack of **Cayce** — an autonomous entity
-[bitcoinhotels.org](https://bitcoinhotels.org) (Bitcoin-native travel guides)
+operating under the Próspera ZEDE agentic-companies jurisdiction pilot. Cayce runs
-and [hardcap.news](https://hardcap.news) (Bitcoin security intelligence).
+[bitcoinhotels.org](https://bitcoinhotels.org) (Bitcoin-native travel guides) and
+[hardcap.news](https://hardcap.news) (Bitcoin security intelligence). The
+experiment is managed by Cayce itself.
 
 Five decoupled components share one Postgres database. Each is a CLI command,
 driven by cron/systemd timers. Nothing runs without an audit trail.

cayceops/components/executor/executor.py

@@ -41,14 +41,20 @@ GATE_ERROR = 2  # missing/invalid configuration or signature
 
 
 def _signing(cfg: Config) -> tuple[str, str]:
-    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode."""
+    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode.
+
+    Also rejects the agent's own identity as operator (D16): the operator
+    key must belong to a human and must never reside on this server.
+    """
     from cayceops.config import ConfigError
+    from cayceops.signing import validate_no_self_approval
 
     if not cfg.operator_pubkey or not cfg.deployment_id:
         raise ConfigError(
             "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and "
             "CAYCE_OPS_DEPLOYMENT_ID are both required"
         )
+    validate_no_self_approval(cfg)
     return cfg.deployment_id, cfg.operator_pubkey
 
 

cayceops/components/maild/maild.py

@@ -43,14 +43,20 @@ GATE_ERROR = 2  # missing/invalid configuration or signature
 
 
 def _signing(cfg: Config) -> tuple[str, str]:
-    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode."""
+    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode.
+
+    Also rejects the agent's own identity as operator (D16): the operator
+    key must belong to a human and must never reside on this server.
+    """
     from cayceops.config import ConfigError
+    from cayceops.signing import validate_no_self_approval
 
     if not cfg.operator_pubkey or not cfg.deployment_id:
         raise ConfigError(
             "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and "
             "CAYCE_OPS_DEPLOYMENT_ID are both required"
         )
+    validate_no_self_approval(cfg)
     return cfg.deployment_id, cfg.operator_pubkey
 
 

cayceops/components/publisher/publisher.py

@@ -118,13 +118,20 @@ GATE_ERROR = 2  # missing/invalid configuration or signature
 
 
 def _signing(cfg: Config) -> tuple[str, str]:
+    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode.
+
+    Also rejects the agent's own identity as operator (D16): the operator
+    key must belong to a human and must never reside on this server.
+    """
     from cayceops.config import ConfigError
+    from cayceops.signing import validate_no_self_approval
 
     if not cfg.operator_pubkey or not cfg.deployment_id:
         raise ConfigError(
             "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and "
             "CAYCE_OPS_DEPLOYMENT_ID are both required"
         )
+    validate_no_self_approval(cfg)
     return cfg.deployment_id, cfg.operator_pubkey
 
 

cayceops/signing.py

@@ -101,3 +101,72 @@ def verify_signature(
         )
     except (ValueError, TypeError):
         return False
+
+
+def pubkey_from_private_hex(private_key_hex: str) -> str:
+    """Derive the 64-char x-only public key from a 32-byte hex private key.
+
+    Raises ValueError on malformed input.
+    """
+    if not private_key_hex:
+        raise ValueError("empty private key")
+    try:
+        from coincurve import PublicKey
+
+        return PublicKey.from_secret(bytes.fromhex(private_key_hex)).format().hex()[2:]
+    except (ImportError, ValueError) as exc:
+        raise ValueError(f"cannot derive pubkey: {exc}") from exc
+
+
+class SelfApprovalError(RuntimeError):
+    """Raised when the configured operator pubkey is the identity of any
+    key held by the system itself (e.g. Cayce nostr identity). The design
+    requires a *separate human* operator key — never an agent key."""
+
+
+FORBIDDEN_IDENTITIES_ENV = "CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS"
+
+
+def forbidden_identities(cfg) -> list[str]:
+    """Pubkeys that may never act as the operator key.
+
+    Includes:
+      1. Any private key loaded from the config (nostr_private_key, or any
+         future agent identity).
+      2. An explicit deny-list in CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS
+         (comma-separated hex pubkeys) for keys held outside this process.
+    """
+    listed: list[str] = []
+    if cfg.nostr_private_key:
+        try:
+            listed.append(pubkey_from_private_hex(cfg.nostr_private_key))
+        except ValueError:
+            pass  # malformed nostr key: _signing() already catches it elsewhere
+    import os
+
+    raw = os.environ.get(FORBIDDEN_IDENTITIES_ENV, "")
+    for tok in raw.split(","):
+        tok = tok.strip().lower()
+        if len(tok) == 64:
+            listed.append(tok)
+    return listed
+
+
+def validate_no_self_approval(cfg) -> None:
+    """Refuse to operate if the operator pubkey equals any identity held by
+    the system. Raises SelfApprovalError.
+
+    Empty operator_pubkey is fine here (that's the 'not configured' state
+    which _signing() rejects elsewhere); this check is about *wrong* keys.
+    """
+    if not cfg.operator_pubkey:
+        return
+    me = cfg.operator_pubkey.strip().lower()
+    for forbidden in forbidden_identities(cfg):
+        if me == forbidden:
+            raise SelfApprovalError(
+                "CAYCE_OPS_OPERATOR_PUBKEY is an agent identity (Cayce or "
+                "another in-process key). The operator key must belong to a "
+                "human and must never reside on this server. Refusing to "
+                "operate. See docs/DECISIONS.md D16."
+            )

docs/DECISIONS.md

@@ -117,3 +117,30 @@ Any negation token anywhere in a mail vetoes an `approval` classification →
 needs_review. Simpler and strictly conservative vs. per-pattern negation
 lists. *Cost:* genuinely-positive approvals containing stray negations
 (e.g. "we have no reason not to welcome you") need human triage.
+
+## D16 — La clave del operador nunca es una identidad de agente
+
+**Fecha:** 2026-10-28
+
+**Contexto:** se descubrió que `CAYCE_OPS_OPERATOR_PUBKEY` estaba configurada
+con la pubkey de Cayce — es decir, Cayce se aprobaba a sí misma. La firma
+BIP-340 era técnicamente válida pero semánticamente nula: "aprobación humana"
+firmada por el agente no es aprobación humana.
+
+**Regla:** los daemons y el CLI se niegan a operar si la pubkey del operador
+coincide con:
+1. La pubkey derivada de cualquier clave privada cargada en la config
+   (`CAYCE_OPS_NOSTR_PRIVATE_KEY` u otra identidad de agente futura), o
+2. Cualquier pubkey en `CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS` (deny-list
+   explícita, para claves fuera de este proceso).
+
+**Implementación:** `cayceops.signing.validate_no_self_approval()`, llamada
+desde `_signing()` en executor/maild/publisher. Lanza `SelfApprovalError`.
+
+**La firma del operador se produce FUERA de la VPS.** El CLI `approve-*`
+acepta la firma como parámetro (ya lo hace); nunca la genera él. Mientras
+no exista el signer externo, `CAYCE_OPS_OPERATOR_PUBKEY` debe permanecer
+vacía — los daemons se niegan a operar y eso es lo correcto.
+
+**La nsec de Cayce NO se toca:** sigue siendo la identidad del agente para
+publicar y anclar. Lo que se prohíbe es usarla como clave de operador.

tests/integration/test_pg_roles.py

@@ -197,7 +197,7 @@ def test_roles_full_daemon_cycle(roles_applied):
     sig = "c9" * 64  # invented 128-hex signature for the permission test
     with _connect_as("cayce_approver") as conn, conn.cursor() as cur:
         cur.execute(
-            f"UPDATE proposals SET status='APPROVED', approved_by='gustavo', "
+            f"UPDATE proposals SET status='APPROVED', approved_by='operator', "
             f"approved_at=now(), operator_signature='{sig}' WHERE id={proposal_id}"
         )
 
@@ -225,7 +225,7 @@ def test_roles_full_daemon_cycle(roles_applied):
     with _connect_as("cayce_approver") as conn, conn.cursor() as cur:
         cur.execute(
             "UPDATE mail_messages SET draft_status='APPROVED', "
-            "approved_by='gustavo', approved_hash='"
+            "approved_by='operator', approved_hash='"
             + "a" * 64
             + f"', operator_signature='{sig}' WHERE id={mail_id}"
         )

tests/test_executor.py

@@ -23,13 +23,13 @@ def test_approve_requires_proposed_state(cfg, sess, signer):
     sess.flush()
 
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
     )
     # second approve must fail (state is now approved)
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 1
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 1
     )
-    assert p.approved_by == "gustavo"
+    assert p.approved_by == "operator"
     assert p.status is ProposalStatus.APPROVED
     # FULL signature stored on the row
     assert p.operator_signature and len(p.operator_signature) == 128
@@ -39,7 +39,7 @@ def test_reject_records_actor(cfg, sess):
     p = _proposal()
     sess.add(p)
     sess.flush()
-    assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
+    assert executor.reject(p.id, "operator", cfg(), sess) == 0
     assert p.status is ProposalStatus.REJECTED
 
 
@@ -50,7 +50,7 @@ def test_run_executes_only_approved(cfg, sess, signer):
     sess.flush()
     assert (
         executor.approve(
-            approved.id, "gustavo", cfg(), sess, signature=signer.proposal(approved)
+            approved.id, "operator", cfg(), sess, signature=signer.proposal(approved)
         )
         == 0
     )
@@ -110,18 +110,18 @@ def test_signed_approval_mandatory_and_content_bound(cfg, sess, signer):
     chash = proposal_content_hash(p.kind, p.title, p.url, p.detail)
 
     # no signature -> refused
-    assert executor.approve(p.id, "gustavo", cfg(), sess, signature=None) == 2
+    assert executor.approve(p.id, "operator", cfg(), sess, signature=None) == 2
     # signature for ANOTHER deployment -> refused (required test)
     other_dep = signer.raw(
         signer.msg("proposal", "other-deploy", p.id, chash)
     )
-    assert executor.approve(p.id, "gustavo", cfg(), sess, signature=other_dep) == 2
+    assert executor.approve(p.id, "operator", cfg(), sess, signature=other_dep) == 2
     # signature over OTHER CONTENT -> refused (required test)
     other_content = signer.raw(signer.msg("proposal", "itest", p.id, "00" * 32))
-    assert executor.approve(p.id, "gustavo", cfg(), sess, signature=other_content) == 2
+    assert executor.approve(p.id, "operator", cfg(), sess, signature=other_content) == 2
     # valid signature -> approved
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
     )
     assert p.status is ProposalStatus.APPROVED
     entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one()
@@ -135,7 +135,7 @@ def test_approve_fails_audited_from_cli_wrapper(cfg, sess):
     sess.add(p)
     sess.flush()
 
-    code = executor.approve(p.id, "gustavo", cfg(), sess, signature=None)
+    code = executor.approve(p.id, "operator", cfg(), sess, signature=None)
     assert code == 2
     entry = sess.query(AuditEntry).filter_by(action="approve", result="error").one()
     assert "signature required" in entry.detail["reason"]
@@ -170,7 +170,7 @@ def test_execution_verifies_signature_against_current_content(cfg, sess, signer,
     sess.add(p)
     sess.flush()
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
     )
 
     p.title = "EDITED after approval"  # content changed under the signature
@@ -191,7 +191,7 @@ def test_failing_plugin_retries_then_fails(cfg, sess, signer, monkeypatch):
     sess.add(p)
     sess.flush()
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
     )
 
     calls = {"n": 0}
@@ -232,7 +232,7 @@ def test_unknown_kind_falls_back_to_log_action(cfg, sess, signer):
     sess.add(p)
     sess.flush()
     assert (
-        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+        executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
     )
 
     assert executor.run(cfg(), sess) == 0

tests/test_maild.py

@@ -77,7 +77,7 @@ def _proposal_with_sent_draft(sess) -> Proposal:
         p.id, "partners@acme.example.org", "[cayce] partnership inquiry", "Hello!", sess
     )
     assert (
-        maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+        maild._approve_draft(mail_id, "operator", sess, signature=_approve_signed(sess, mail_id))
         == 0
     )
     assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
@@ -243,7 +243,7 @@ def test_send_requires_explicit_human_approval(cfg, sess):
 
     assert (
         maild._approve_draft(
-            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
         )
         == 0
     )
@@ -277,7 +277,7 @@ def test_send_requires_explicit_human_approval(cfg, sess):
         .order_by(AuditEntry.id.desc())
         .first()
     )
-    assert row is not None and "gustavo" in str(row.detail)
+    assert row is not None and "operator" in str(row.detail)
 
 
 def test_double_approve_is_rejected(cfg, sess):
@@ -288,11 +288,11 @@ def test_double_approve_is_rejected(cfg, sess):
 
     assert (
         maild._approve_draft(
-            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
         )
         == 0
     )
-    assert maild._approve_draft(inbound.id, "gustavo", sess) == 1
+    assert maild._approve_draft(inbound.id, "operator", sess) == 1
 
 
 # --- audit corrections ----------------------------------------------------------
@@ -311,7 +311,9 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
             p.id, f"desk{i}@acme.example.org", f"[cayce] inquiry {i}", "Hello!", sess
         )
         assert (
-            maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+            maild._approve_draft(
+                mail_id, "operator", sess, signature=_approve_signed(sess, mail_id)
+            )
             == 0
         )
         assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
@@ -331,7 +333,7 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
         )
         assert (
             maild._approve_draft(
-                inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+                inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
             )
             == 0
         )
@@ -368,7 +370,7 @@ def test_edited_draft_refuses_to_send(cfg, sess):
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
     assert (
         maild._approve_draft(
-            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
         )
         == 0
     )
@@ -402,7 +404,7 @@ def test_compose_creates_first_contact_draft(cfg, sess):
 
     # approve + send -> an inbound reply can now thread to this proposal
     assert (
-        maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+        maild._approve_draft(mail_id, "operator", sess, signature=_approve_signed(sess, mail_id))
         == 0
     )
     stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None)
@@ -462,7 +464,7 @@ def test_changed_to_addr_after_approval_blocks_send(cfg, sess):
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
     assert (
         maild._approve_draft(
-            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
         )
         == 0
     )
@@ -491,14 +493,14 @@ def test_foreign_deployment_signature_refused(cfg, sess):
     other_dep = _sign(
         OPERATOR_SECRET, approval_message("mail", "other-deploy", inbound.id, chash)
     )
-    assert maild._approve_draft(inbound.id, "gustavo", sess, signature=other_dep) == 2
+    assert maild._approve_draft(inbound.id, "operator", sess, signature=other_dep) == 2
     assert inbound.draft_status is DraftStatus.DRAFTED
 
     # signature valid for OTHER CONTENT
     other_content = _sign(
         OPERATOR_SECRET, approval_message("mail", "itest", inbound.id, "00" * 32)
     )
-    assert maild._approve_draft(inbound.id, "gustavo", sess, signature=other_content) == 2
+    assert maild._approve_draft(inbound.id, "operator", sess, signature=other_content) == 2
     assert inbound.draft_status is DraftStatus.DRAFTED
 
 
@@ -512,7 +514,7 @@ def test_send_failures_audited_and_exit_code(cfg, sess, monkeypatch):
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
     assert (
         maild._approve_draft(
-            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            inbound.id, "operator", sess, signature=_approve_signed(sess, inbound.id)
         )
         == 0
     )

tests/test_models.py

@@ -37,7 +37,7 @@ def test_proposal_status_transitions(sess):
     sess.flush()
 
     p.status = ProposalStatus.APPROVED
-    p.approved_by = "gustavo"
+    p.approved_by = "operator"
     p.approved_at = datetime.now(UTC)
     p.status = ProposalStatus.RUNNING
     p.status = ProposalStatus.DONE
@@ -45,7 +45,7 @@ def test_proposal_status_transitions(sess):
 
     loaded = sess.scalar(select(Proposal).where(Proposal.id == p.id))
     assert loaded.status == ProposalStatus.DONE
-    assert loaded.approved_by == "gustavo"
+    assert loaded.approved_by == "operator"
 
 
 def test_dedupe_key_unique(sess):
@@ -65,12 +65,12 @@ def test_audit_helper_writes(sess):
     p = Proposal(kind="other", title="x")
     sess.add(p)
     sess.flush()
-    audit(sess, actor="gustavo", action="approve", proposal_id=p.id, detail={"note": "ok"})
+    audit(sess, actor="operator", action="approve", proposal_id=p.id, detail={"note": "ok"})
     sess.flush()
 
     rows = list(sess.scalars(select(AuditEntry)).all())
     assert len(rows) == 1
-    assert rows[0].actor == "gustavo"
+    assert rows[0].actor == "operator"
     assert rows[0].proposal_id == p.id
 
 

tests/test_no_self_approval.py

@@ -0,0 +1,82 @@
+"""Tests for D16: the operator key may never be an agent identity.
+
+Background: the operator key was, at some point, Cayce's nostr key. The
+daemon must refuse to operate in that configuration, because "human
+approval" signed by the agent is not human approval.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from cayceops.signing import (
+    SelfApprovalError,
+    pubkey_from_private_hex,
+    validate_no_self_approval,
+)
+
+# keys invented for tests only (allowlisted in .gitleaks.toml)
+CAYCE_SECRET = "11" * 32  # simulates Cayce nostr identity
+CAYCE_PUBKEY = pubkey_from_private_hex(CAYCE_SECRET)
+HUMAN_SECRET = "33" * 32  # a proper human operator key
+HUMAN_PUBKEY = pubkey_from_private_hex(HUMAN_SECRET)
+
+
+def _cfg(monkeypatch, operator_pubkey: str, nostr_private_key: str = CAYCE_SECRET):
+    from cayceops.config import Config
+
+    monkeypatch.setenv("CAYCE_OPS_DEPLOYMENT_ID", "itest")
+    monkeypatch.setenv("CAYCE_OPS_OPERATOR_PUBKEY", operator_pubkey)
+    monkeypatch.setenv("CAYCE_OPS_NOSTR_PRIVATE_KEY", nostr_private_key)
+    return Config()
+
+
+def test_agent_pubkey_as_operator_is_refused(monkeypatch):
+    """The exact bug: operator pubkey = Cayce pubkey must fail."""
+    cfg = _cfg(monkeypatch, operator_pubkey=CAYCE_PUBKEY)
+    with pytest.raises(SelfApprovalError):
+        validate_no_self_approval(cfg)
+
+
+def test_human_pubkey_passes(monkeypatch):
+    cfg = _cfg(monkeypatch, operator_pubkey=HUMAN_PUBKEY)
+    validate_no_self_approval(cfg)  # does not raise
+
+
+def test_empty_operator_pubkey_passes_validation(monkeypatch):
+    """Empty is the 'not configured' state — rejected elsewhere, not here."""
+    cfg = _cfg(monkeypatch, operator_pubkey="")
+    validate_no_self_approval(cfg)
+
+
+def test_explicit_denylist_is_honoured(monkeypatch):
+    """A key held outside this process can be listed explicitly."""
+    outside_pubkey = pubkey_from_private_hex("44" * 32)
+    monkeypatch.setenv(
+        "CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS", f"{outside_pubkey}, {HUMAN_PUBKEY}"
+    )
+    cfg = _cfg(monkeypatch, operator_pubkey=HUMAN_PUBKEY, nostr_private_key="")
+    with pytest.raises(SelfApprovalError):
+        validate_no_self_approval(cfg)
+
+
+def test_daemon_refuses_to_start_with_agent_key(monkeypatch, sess, db_engine):
+    """End-to-end: executor.run must refuse (exit 2) when the operator
+    pubkey is Cayce's identity."""
+    from cayceops.components.executor import executor
+    from cayceops.models import DaemonRun
+
+    cfg = _cfg(monkeypatch, operator_pubkey=CAYCE_PUBKEY)
+    code = executor.run(cfg, sess)
+    assert code == 2
+    run_row = sess.query(DaemonRun).filter_by(daemon="executor").one()
+    assert run_row.status == "error"
+    assert "agent identity" in run_row.detail["refused"]
+
+
+def test_pubkey_derivation_roundtrip():
+    assert pubkey_from_private_hex(HUMAN_SECRET) == HUMAN_PUBKEY
+    with pytest.raises(ValueError):
+        pubkey_from_private_hex("")
+    with pytest.raises(ValueError):
+        pubkey_from_private_hex("nothex")

tests/test_publisher.py

@@ -44,7 +44,7 @@ def _approved_item(cfg, sess, **kw) -> ContentItem:
     item = _item(status=ContentStatus.DRAFT, **kw)
     sess.add(item)
     sess.flush()
-    assert publisher._approve(item.id, "gustavo", sess, signature=_sign_content(item)) == 0
+    assert publisher._approve(item.id, "operator", sess, signature=_sign_content(item)) == 0
     return item
 
 
@@ -92,16 +92,16 @@ def test_approve_gate(cfg, sess):
     cfg()  # signing env first
 
     # unsigned approval refused (mandatory signature, no degraded mode)
-    assert publisher._approve(item.id, "gustavo", sess, signature=None) == 2
+    assert publisher._approve(item.id, "operator", sess, signature=None) == 2
     # signed approval accepted, FULL signature stored
     sig = _sign_content(item)
-    assert publisher._approve(item.id, "gustavo", sess, signature=sig) == 0
+    assert publisher._approve(item.id, "operator", sess, signature=sig) == 0
     assert item.status is ContentStatus.APPROVED
     assert item.operator_signature == sig and len(sig) == 128
-    assert publisher._approve(item.id, "gustavo", sess, signature=sig) == 1  # not draft anymore
+    assert publisher._approve(item.id, "operator", sess, signature=sig) == 1  # not draft anymore
 
     actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor
-    assert actor == "gustavo"
+    assert actor == "operator"
 
 
 # --- publishing ------------------------------------------------------------------
@@ -183,7 +183,7 @@ def test_foreign_deployment_signature_refused(cfg, sess, signer):
     other_dep = signer.raw(
         signer.msg("content", "other-deploy", item.id, signer.content_hash(item))
     )
-    assert publisher._approve(item.id, "gustavo", sess, signature=other_dep) == 2
+    assert publisher._approve(item.id, "operator", sess, signature=other_dep) == 2
     assert item.status is ContentStatus.DRAFT