audit-1 (critical 1): mandatory operator signatures bound to content+deployment

- signing module: canonical content hashes + approval_message
  sha256(cayce-ops:<type>:<deployment_id>:<id>:<content_hash>)
- approve/approve-draft/publisher-approve REQUIRE signature; without
  CAYCE_OPS_OPERATOR_PUBKEY+DEPLOYMENT_ID the daemons refuse to
  execute/send/publish (no degraded mode)
- run-time re-verification at execute/send/publish: unsigned rows (raw SQL
  approval) or edited content revert to pre-approval state, audited as error
- FULL signatures stored on rows (operator_signature, migration 0003) and
  in audit detail — never truncated
- signature failures audited from CLI wrappers too (1f)
- AUTONOMOUS_MODE approval path removed (D13): no non-human approvals

tests: SQL-approved row refused at execute/send/publish; to_addr change
blocks send; foreign deployment/content signatures refused; full-signature
persistence. 65 tests green.

tests/conftest.py

@@ -10,8 +10,14 @@ from __future__ import annotations
 
 import os
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
+from coincurve import PublicKey
+
+# test operator keypair (invented; allowlisted in .gitleaks.toml)
+OPERATOR_SECRET = bytes.fromhex("22" * 32)
+OPERATOR_PUBKEY = PublicKey.from_secret(OPERATOR_SECRET).format().hex()[2:]
 
 # Config must see the test DB before cayceops.config/db get imported.
 os.environ.setdefault("CAYCE_OPS_DATABASE_URL", "sqlite:///:memory:")
@@ -54,6 +60,61 @@ def sess(db_engine):
     s.close()
 
 
+@pytest.fixture()
+def signer():
+    """Signs approval messages with the test operator key (22*32, deployment
+    'itest'). Mirrors exactly what the operator CLI does."""
+    from coincurve import PrivateKey
+
+    from cayceops.models import MailDirection
+    from cayceops.signing import (
+        approval_message,
+        content_content_hash,
+        mail_content_hash,
+        proposal_content_hash,
+    )
+
+    secret = OPERATOR_SECRET
+
+    def _sig(msg: bytes) -> str:
+        return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()
+
+    def proposal(p, deployment="itest"):
+        chash = proposal_content_hash(p.kind, p.title, p.url, p.detail)
+        return _sig(approval_message("proposal", deployment, p.id, chash))
+
+    def mail(record, deployment="itest"):
+        to = (
+            record.from_addr
+            if record.direction == MailDirection.INBOUND
+            else record.to_addr
+        )
+        chash = mail_content_hash(to, record.subject, record.draft_reply)
+        return _sig(approval_message("mail", deployment, record.id, chash))
+
+    def content(item, deployment="itest"):
+        chash = content_content_hash(item.channel, item.caption, item.link)
+        return _sig(approval_message("content", deployment, item.id, chash))
+
+    def raw(sig_bytes: bytes) -> str:
+        return _sig(sig_bytes)
+
+    def msg(item_type: str, deployment: str, item_id: int, chash: str) -> bytes:
+        from cayceops.signing import approval_message
+
+        return approval_message(item_type, deployment, item_id, chash)
+
+    def content_hash(item, deployment="itest") -> str:
+        from cayceops.signing import content_content_hash
+
+        return content_content_hash(item.channel, item.caption, item.link)
+
+    return SimpleNamespace(
+        proposal=proposal, mail=mail, content=content, raw=raw, msg=msg,
+        content_hash=content_hash,
+    )
+
+
 @pytest.fixture()
 def cfg(monkeypatch):
     """A Config snapshot with test defaults (no network, no autonomy)."""
@@ -66,6 +127,10 @@ def cfg(monkeypatch):
         "CAYCE_OPS_ENABLE_INSTAGRAM": "false",
         "CAYCE_OPS_NOSTR_RELAYS": "",
         "CAYCE_OPS_MEDIA_DIR": "media",
+        # signing defaults (audit round 2): tests operate in a signed
+        # deployment unless they explicitly clear these
+        "CAYCE_OPS_DEPLOYMENT_ID": "itest",
+        "CAYCE_OPS_OPERATOR_PUBKEY": OPERATOR_PUBKEY,
     }
 
     def _apply(**overrides):

tests/test_executor.py

@@ -1,4 +1,9 @@
-"""executor tests — human gate, autonomy bounds, retries, audit chain."""
+"""executor tests — human gate (signed), autonomy bounds, retries, audit chain.
+
+Since audit round 2, approvals REQUIRE an operator signature bound to the
+proposal's current content + deployment id (tests sign with the conftest
+operator key 22*32, deployment 'itest').
+"""
 
 from __future__ import annotations
 
@@ -12,16 +17,22 @@ def _proposal(kind="content_gap", title="Test item", **kw):
     return Proposal(kind=kind, title=title, **kw)
 
 
-def test_approve_requires_proposed_state(cfg, sess):
+def test_approve_requires_proposed_state(cfg, sess, signer):
     p = _proposal()
     sess.add(p)
     sess.flush()
 
-    assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+    )
     # second approve must fail (state is now approved)
-    assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 1
+    )
     assert p.approved_by == "gustavo"
-    assert p.status == ProposalStatus.APPROVED
+    assert p.status is ProposalStatus.APPROVED
+    # FULL signature stored on the row
+    assert p.operator_signature and len(p.operator_signature) == 128
 
 
 def test_reject_records_actor(cfg, sess):
@@ -29,75 +40,26 @@ def test_reject_records_actor(cfg, sess):
     sess.add(p)
     sess.flush()
     assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
-    assert p.status == ProposalStatus.REJECTED
+    assert p.status is ProposalStatus.REJECTED
 
 
-def test_signed_approval_required_when_operator_key_configured(cfg, sess):
-    """Critical finding 1: --actor was free text; now a key can be required."""
-    from coincurve import PrivateKey, PublicKey
-
-    from cayceops.components.executor import executor
-
-    secret = bytes.fromhex("22" * 32)
-    pubkey = PublicKey.from_secret(secret).format().hex()[2:]
-    p = _proposal()
-    sess.add(p)
-    sess.flush()
-
-    config = cfg(CAYCE_OPS_OPERATOR_PUBKEY=pubkey)
-
-    # no signature -> refused
-    assert executor.approve(p.id, "gustavo", config, sess, signature=None) == 2
-    # wrong key -> refused
-    bad = (
-        PrivateKey(bytes.fromhex("33" * 32))
-        .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32)
-        .hex()
-    )
-    assert executor.approve(p.id, "gustavo", config, sess, signature=bad) == 2
-    # correct key -> approved
-    good = (
-        PrivateKey(secret)
-        .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32)
-        .hex()
-    )
-    assert executor.approve(p.id, "gustavo", config, sess, signature=good) == 0
-    assert p.status is ProposalStatus.APPROVED
-    entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one()
-    assert "signature" in entry.detail
-
-
-def test_auto_approve_orders_by_relevance_desc(cfg, sess):
-    """Medium finding: ascending 'risk_score' approved noise first."""
-    low = _proposal(kind="content_gap", title="barely relevant")
-    low.match_score = 0.1
-    high = _proposal(kind="content_gap", title="highly relevant")
-    high.match_score = 0.9
-    sess.add_all([low, high])
-    sess.flush()
-
-    config = cfg(
-        CAYCE_OPS_AUTONOMOUS_MODE="true",
-        CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
-        CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
-    )
-    executor._auto_approve(config, sess)
-    assert high.status is ProposalStatus.APPROVED
-    assert low.status is ProposalStatus.PROPOSED
-
-
-def test_run_executes_only_approved(cfg, sess):
+def test_run_executes_only_approved(cfg, sess, signer):
     approved = _proposal(title="Approved item")
     proposed = _proposal(title="Proposed item")
     sess.add_all([approved, proposed])
     sess.flush()
-    executor.approve(approved.id, "gustavo", cfg(), sess)
+    assert (
+        executor.approve(
+            approved.id, "gustavo", cfg(), sess, signature=signer.proposal(approved)
+        )
+        == 0
+    )
 
     config = cfg()  # AUTONOMOUS_MODE=false
     assert executor.run(config, sess) == 0
 
-    assert approved.status == ProposalStatus.DONE
-    assert proposed.status == ProposalStatus.PROPOSED  # untouched
+    assert approved.status is ProposalStatus.DONE
+    assert proposed.status is ProposalStatus.PROPOSED  # untouched
 
     acts = [a.action for a in sess.query(AuditEntry).order_by(AuditEntry.id).all()]
     assert "execute_start" in acts and "execute_done" in acts
@@ -110,36 +72,127 @@ def test_autonomy_off_by_default(cfg, sess):
     sess.flush()
     config = cfg()  # autonomous_mode False even with kinds configured
     executor._auto_approve(config, sess)
-    assert p.status == ProposalStatus.PROPOSED
+    assert p.status is ProposalStatus.PROPOSED
 
 
-def test_autonomy_respects_kinds_and_daily_cap(cfg, sess):
-    p1 = _proposal(kind="content_gap", title="auto ok")
-    p2 = _proposal(kind="content_gap", title="auto over cap")
-    p3 = _proposal(kind="affiliate", title="kind not allowed")
-    sess.add_all([p1, p2, p3])
+def test_autonomy_cannot_approve_anymore(cfg, sess):
+    """Round-2 1a: with mandatory signatures there is NO non-human approval
+    path. AUTONOMOUS_MODE no longer approves anything (D13)."""
+    low = _proposal(kind="content_gap", title="barely relevant")
+    low.match_score = 0.1
+    high = _proposal(kind="content_gap", title="highly relevant")
+    high.match_score = 0.9
+    sess.add_all([low, high])
     sess.flush()
-
     config = cfg(
         CAYCE_OPS_AUTONOMOUS_MODE="true",
         CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
         CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
     )
-    approved = executor._auto_approve(config, sess)
-    assert approved == 1
-    assert p1.status == ProposalStatus.APPROVED
-    assert p2.status == ProposalStatus.PROPOSED  # cap reached
-    assert p3.status == ProposalStatus.PROPOSED  # kind not in policy
-
-    actor = sess.query(AuditEntry).filter_by(action="approve").one().actor
-    assert actor == "auto:policy"
+    assert executor._auto_approve(config, sess) == 0
+    assert low.status is ProposalStatus.PROPOSED
+    assert high.status is ProposalStatus.PROPOSED
+    entry = sess.query(AuditEntry).filter_by(action="autonomy_disabled").one()
+    assert entry.result == "error"
 
 
-def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
+def test_signed_approval_mandatory_and_content_bound(cfg, sess, signer):
+    """Round-2 1a/1b/1c: signature mandatory, bound to content+deployment,
+    stored COMPLETE on the row and in the audit log. Signatures for other
+    deployments or other content are refused."""
+    from cayceops.signing import proposal_content_hash
+
+    p = _proposal(kind="partnership", title="Acme partner", url="https://acme.example.org")
+    p.detail = {"region": "latam"}
+    sess.add(p)
+    sess.flush()
+
+    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
+    # 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
+    # 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
+    # valid signature -> approved
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+    )
+    assert p.status is ProposalStatus.APPROVED
+    entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one()
+    assert entry.detail["signature"] == p.operator_signature
+
+
+def test_approve_fails_audited_from_cli_wrapper(cfg, sess):
+    """Round-2 1f: signature failures leave an audit record even when the
+    command runs through its CLI wrapper (the wrapper passes sess)."""
+    p = _proposal()
+    sess.add(p)
+    sess.flush()
+
+    code = executor.approve(p.id, "gustavo", 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"]
+
+
+def test_sql_approved_row_is_refused_at_execution(cfg, sess, monkeypatch):
+    """Round-2 1d (required test): a row put in `approved` by raw SQL without
+    a signature must not execute; it reverts to `proposed` with an error audit."""
+    p = _proposal()
+    p.status = ProposalStatus.APPROVED  # raw-SQL style: no signature
+    sess.add(p)
+    sess.flush()
+
+    executed = []
+
+    def spy_plugin(proposal, cfg, sess):
+        executed.append(proposal.id)
+        return {}
+
+    monkeypatch.setitem(plugins.REGISTRY, "content_gap", spy_plugin)
+    assert executor.run(cfg(), sess) == 0
+    assert executed == []
+    assert p.status is ProposalStatus.PROPOSED  # reverted
+    refused = sess.query(AuditEntry).filter_by(action="execute_refused").one()
+    assert refused.result == "error"
+
+
+def test_execution_verifies_signature_against_current_content(cfg, sess, signer, monkeypatch):
+    """Round-2 1d: signature over content that changed after approval is
+    refused at execution time."""
+    p = _proposal(kind="partnership", title="original")
+    sess.add(p)
+    sess.flush()
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+    )
+
+    p.title = "EDITED after approval"  # content changed under the signature
+
+    executed = []
+    monkeypatch.setitem(
+        plugins.REGISTRY,
+        "partnership",
+        lambda proposal, cfg, sess: executed.append(proposal.id) or {},
+    )
+    assert executor.run(cfg(), sess) == 0
+    assert executed == []
+    assert p.status is ProposalStatus.PROPOSED
+
+
+def test_failing_plugin_retries_then_fails(cfg, sess, signer, monkeypatch):
     p = _proposal(kind="affiliate", title="Will fail")
     sess.add(p)
     sess.flush()
-    executor.approve(p.id, "gustavo", cfg(), sess)
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+    )
 
     calls = {"n": 0}
 
@@ -151,11 +204,11 @@ def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
     config = cfg(CAYCE_OPS_MAX_ATTEMPTS="2")
 
     executor.run(config, sess)  # attempt 1 -> back to approved for retry
-    assert p.status == ProposalStatus.APPROVED
+    assert p.status is ProposalStatus.APPROVED
     assert p.attempts == 1
 
     executor.run(config, sess)  # attempt 2 -> failed for good
-    assert p.status == ProposalStatus.FAILED
+    assert p.status is ProposalStatus.FAILED
     assert p.attempts == 2
     assert calls["n"] == 2
 
@@ -174,14 +227,16 @@ def test_http_post_blocked_when_disabled(cfg, sess):
         plugins.http_post(p, config, sess)
 
 
-def test_unknown_kind_falls_back_to_log_action(cfg, sess):
+def test_unknown_kind_falls_back_to_log_action(cfg, sess, signer):
     p = _proposal(kind="brand_new_kind", title="No plugin yet")
     sess.add(p)
     sess.flush()
-    executor.approve(p.id, "gustavo", cfg(), sess)
+    assert (
+        executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0
+    )
 
     assert executor.run(cfg(), sess) == 0
-    assert p.status == ProposalStatus.DONE
+    assert p.status is ProposalStatus.DONE
 
     done = sess.query(AuditEntry).filter_by(action="execute_done").one()
     assert done.detail["outcome"]["plugin_fallback"] is True

tests/test_maild.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
-from types import SimpleNamespace
+import os
+
+from coincurve import PrivateKey
+from coincurve import PublicKey as _PK
 
 from cayceops.components.maild import classify as clf
 from cayceops.components.maild import maild
@@ -16,9 +19,20 @@ from cayceops.models import (
     ProposalStatus,
 )
 
+# test operator keypair (matches conftest default pubkey; invented)
+OPERATOR_SECRET = bytes.fromhex("22" * 32)
+OPERATOR_PUBKEY = _PK.from_secret(OPERATOR_SECRET).format().hex()[2:]
+
 OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
 
-STUB_CFG = SimpleNamespace(mail_from="agent@cayce-co.xyz")
+# the signing gate reads env at runtime — make sure every path in this module
+# (fixtures included) sees a signed deployment
+os.environ.setdefault("CAYCE_OPS_DEPLOYMENT_ID", "itest")
+os.environ.setdefault("CAYCE_OPS_OPERATOR_PUBKEY", OPERATOR_PUBKEY)
+from cayceops.config import load as _load  # noqa: E402  (env set above)
+
+STUB_CFG = _load()  # a real, signed Config (replaces the old SimpleNamespace)
+
 
 
 def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes:
@@ -34,9 +48,24 @@ def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes:
     return "\r\n".join(lines).encode()
 
 
+def _sign(msg: bytes, secret: bytes = OPERATOR_SECRET) -> str:
+
+    return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()
+
+
+def _approve_signed(sess, mail_id: int) -> str:
+    """Approve with a valid operator signature over the row's current content."""
+    from cayceops.signing import approval_message, mail_content_hash
+
+    record = sess.get(MailMessage, mail_id)
+    to_addr = record.from_addr if record.direction == MailDirection.INBOUND else record.to_addr
+    chash = mail_content_hash(to_addr, record.subject, record.draft_reply)
+    return _sign(approval_message("mail", "itest", mail_id, chash))
+
+
 def _proposal_with_sent_draft(sess) -> Proposal:
     """A proposal whose first-contact mail went through the REAL path:
-    compose -> approve -> send (creating a true OUTBOUND row)."""
+    compose -> approve (signed) -> send (creating a true OUTBOUND row)."""
     p = Proposal(
         kind="partnership",
         title="Acme hotel affiliate program",
@@ -47,7 +76,10 @@ def _proposal_with_sent_draft(sess) -> Proposal:
     mail_id = maild._compose(
         p.id, "partners@acme.example.org", "[cayce] partnership inquiry", "Hello!", sess
     )
-    assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+    assert (
+        maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+        == 0
+    )
     assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
     out = (
         sess.query(MailMessage)
@@ -72,8 +104,15 @@ def test_rules_classification(cfg):
         clf.classify("Your application was approved, welcome to the program", config)[0]
         is MailClassification.APPROVAL
     )
+    # negation present -> approval is vetoed to needs_review (round-2 rule;
+    # the human decides between rejection and other)
     assert (
         clf.classify("Unfortunately we cannot accept new partners", config)[0]
+        is MailClassification.NEEDS_REVIEW
+    )
+    # plain rejection (no approval word involved) still classifies directly
+    assert (
+        clf.classify("We regret to inform you that your application was declined", config)[0]
         is MailClassification.REJECTION
     )
     assert (
@@ -84,10 +123,16 @@ def test_rules_classification(cfg):
     assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
 
 
-def test_negated_approvals_are_rejections(cfg):
-    """Critical finding 4: 'cannot accept'/'not approved' read as approvals."""
+def test_negated_approvals_never_classify_approval(cfg):
+    """Round-2 point 2: any negation vetoes `approval` -> needs_review."""
     config = cfg()
     cases = [
+        "We cannot currently accept new affiliates",
+        "We will never accept crypto businesses",
+        "Approval was not granted for your site",
+        "We are no longer accepting applications",
+        "Your application hasn't been approved",
+        "We have decided not to move forward with your application",
         "We cannot accept your application",
         "We are not accepting new partners at this time",
         "Your application was not approved",
@@ -95,9 +140,8 @@ def test_negated_approvals_are_rejections(cfg):
         "We do not accept external affiliates",
     ]
     for text in cases:
-        kind, confidence = clf.classify(text, config)
-        assert kind is MailClassification.REJECTION, f"{text!r} -> {kind}"
-        assert confidence >= 0.9
+        kind, _ = clf.classify(text, config)
+        assert kind is not MailClassification.APPROVAL, f"{text!r} -> {kind}"
 
 
 def test_llm_prompt_format_survives_json_braces():
@@ -197,7 +241,12 @@ def test_send_requires_explicit_human_approval(cfg, sess):
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
     original_id = inbound.message_id
 
-    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+    assert (
+        maild._approve_draft(
+            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+        )
+        == 0
+    )
 
     sent = []
     config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
@@ -237,7 +286,12 @@ def test_double_approve_is_rejected(cfg, sess):
     maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
 
-    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+    assert (
+        maild._approve_draft(
+            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+        )
+        == 0
+    )
     assert maild._approve_draft(inbound.id, "gustavo", sess) == 1
 
 
@@ -256,7 +310,10 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
         mail_id = maild._compose(
             p.id, f"desk{i}@acme.example.org", f"[cayce] inquiry {i}", "Hello!", sess
         )
-        assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+        assert (
+            maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+            == 0
+        )
         assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
         sess.flush()
         out = sess.get(MailMessage, mail_id)
@@ -272,7 +329,12 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
             .filter_by(direction=MailDirection.INBOUND, message_id=f"<acme-{i}@acme.example.org>")
             .one()
         )
-        maild._approve_draft(inbound.id, "gustavo", sess)
+        assert (
+            maild._approve_draft(
+                inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+            )
+            == 0
+        )
     sess.commit()  # per-message factory sessions only see committed rows
 
     calls = {"n": 0}
@@ -304,7 +366,12 @@ def test_edited_draft_refuses_to_send(cfg, sess):
     raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
     maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
     inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
-    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+    assert (
+        maild._approve_draft(
+            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+        )
+        == 0
+    )
 
     inbound.draft_reply = "TOTALLY EDITED TEXT after approval"  # tamper
 
@@ -314,7 +381,7 @@ def test_edited_draft_refuses_to_send(cfg, sess):
     assert stats["sent"] == 0 and stats["hash_mismatch"] == 1
     assert sent == []
     assert inbound.draft_status is DraftStatus.DRAFTED  # needs fresh approval
-    action = sess.query(AuditEntry).filter_by(action="send_refused_hash_mismatch").one()
+    action = sess.query(AuditEntry).filter_by(action="send_refused").one()
     assert action.result == "error"
 
 
@@ -334,7 +401,10 @@ def test_compose_creates_first_contact_draft(cfg, sess):
     assert out.proposal_id == p.id
 
     # approve + send -> an inbound reply can now thread to this proposal
-    assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+    assert (
+        maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id))
+        == 0
+    )
     stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None)
     assert stats["sent"] == 1
     sess.flush()
@@ -349,3 +419,112 @@ def test_mark_seen_returned_for_post_commit_processing(cfg, sess):
     stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
     assert stats["processed"] == 1
     assert nums == [b"0"]  # caller marks seen only after commit
+
+
+# --- round-2 signature gate (mail) ------------------------------------------------
+
+
+def _approved_reply(sess, cfg) -> MailMessage:
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    return sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+
+
+def test_unsigned_sql_approved_draft_is_refused(cfg, sess):
+    """Round-2 required test: raw-SQL approved row (no signature) not sent."""
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    # raw-SQL style approval: status flips, no signature
+    sess.execute(
+        MailMessage.__table__.update()
+        .where(MailMessage.id == inbound.id)
+        .values(draft_status=DraftStatus.APPROVED)
+    )
+    sess.expire(inbound)
+
+    sent = []
+    stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 0 and stats["hash_mismatch"] == 1
+    assert sent == []
+    assert inbound.draft_status is DraftStatus.DRAFTED  # reverted
+    refused = sess.query(AuditEntry).filter_by(action="send_refused").one()
+    assert refused.result == "error"
+
+
+def test_changed_to_addr_after_approval_blocks_send(cfg, sess):
+    """Round-2 required test: to_addr edited after approval -> refused."""
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert (
+        maild._approve_draft(
+            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+        )
+        == 0
+    )
+
+    inbound.from_addr = "attacker@evil.example.org"  # redirect target
+
+    sent = []
+    stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 0 and stats["hash_mismatch"] == 1
+    assert sent == []
+    assert inbound.draft_status is DraftStatus.DRAFTED
+
+
+def test_foreign_deployment_signature_refused(cfg, sess):
+    """Round-2 required test: a valid signature for another deployment_id (or
+    other content) is refused."""
+    from cayceops.signing import approval_message, mail_content_hash
+
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    chash = mail_content_hash(inbound.from_addr, inbound.subject, inbound.draft_reply)
+
+    # signature valid for ANOTHER deployment
+    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 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 inbound.draft_status is DraftStatus.DRAFTED
+
+
+def test_send_failures_audited_and_exit_code(cfg, sess, monkeypatch):
+    """Round-2 point 5: failures audited + surfaced; non-zero exit handled by CLI."""
+    import cayceops.db as appdb
+
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert (
+        maild._approve_draft(
+            inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id)
+        )
+        == 0
+    )
+
+    def boom(cfg, to, subject, body, message_id, in_reply_to=None):
+        raise OSError("relay down")
+
+
+    sess.commit()  # per-message factory sessions only see committed rows
+    stats = maild._send_approved(
+        STUB_CFG, sess, sender=boom, factory=appdb.session_factory()
+    )
+    assert stats["error"] == 1 and stats["sent"] == 0
+    failed = sess.query(AuditEntry).filter_by(action="send_failed").one()
+    assert failed.result == "error"

tests/test_models.py

@@ -110,3 +110,35 @@ def test_audit_chain_and_verification(sess):
     first.actor = "tampered"
     ok, broken = verify_audit_chain(sess)
     assert not ok and broken == first.id
+
+
+def test_concurrent_writers_keep_chain_verifiable(db_engine):
+    """Round-2 point 4: serialized writes -> no forked chains."""
+    import threading
+
+    import cayceops.db as appdb
+
+    errors: list[Exception] = []
+
+    def writer(worker: int, n: int):
+        try:
+            for i in range(n):
+                with appdb.session_scope() as s:
+                    audit(s, actor=f"w{worker}", action=f"step{i}", detail={"worker": worker})
+        except Exception as exc:  # noqa: BLE001
+            errors.append(exc)
+
+    threads = [threading.Thread(target=writer, args=(w, 25)) for w in range(4)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert not errors, errors
+    s = appdb.session_factory()()
+    total = s.query(AuditEntry).count()
+    assert total == 100
+    from cayceops.models import verify_audit_chain
+
+    ok, broken = verify_audit_chain(s)
+    assert ok and broken is None

tests/test_publisher.py

@@ -1,16 +1,30 @@
-"""publisher tests — calendar, human gate, nostr publish, disabled adapters."""
+"""publisher tests — calendar, human gate (signed), nostr publish, gates."""
 
 from __future__ import annotations
 
-import os
+from datetime import UTC, datetime, timedelta
 
 import pytest
 
 from cayceops.components.publisher import adapters, nostr, publisher
 from cayceops.models import AuditEntry, ContentItem, ContentStatus
 
-# invented test key (never a real nostr key — allowlisted in .gitleaks.toml)
-TEST_KEY = "11" * 32
+# invented test keys (never real — allowlisted in .gitleaks.toml)
+TEST_KEY = "11" * 32  # nostr signing key
+OPERATOR_SECRET = bytes.fromhex("22" * 32)  # approval signing key
+
+
+def _sign(msg: bytes, secret: bytes = OPERATOR_SECRET) -> str:
+    from coincurve import PrivateKey
+
+    return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()
+
+
+def _sign_content(item: ContentItem, deployment: str = "itest") -> str:
+    from cayceops.signing import approval_message, content_content_hash
+
+    chash = content_content_hash(item.channel, item.caption, item.link)
+    return _sign(approval_message("content", deployment, item.id, chash))
 
 
 def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED):
@@ -24,6 +38,16 @@ def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatu
     )
 
 
+def _approved_item(cfg, sess, **kw) -> ContentItem:
+    """A content item properly approved with a signature (env set first)."""
+    cfg()  # sets the signing env before any approval
+    item = _item(status=ContentStatus.DRAFT, **kw)
+    sess.add(item)
+    sess.flush()
+    assert publisher._approve(item.id, "gustavo", sess, signature=_sign_content(item)) == 0
+    return item
+
+
 # --- nostr client -------------------------------------------------------------
 
 
@@ -65,10 +89,16 @@ def test_approve_gate(cfg, sess):
     item = _item(status=ContentStatus.DRAFT)
     sess.add(item)
     sess.flush()
+    cfg()  # signing env first
 
-    assert publisher._approve(item.id, "gustavo", sess) == 0
+    # unsigned approval refused (mandatory signature, no degraded mode)
+    assert publisher._approve(item.id, "gustavo", 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 item.status is ContentStatus.APPROVED
-    assert publisher._approve(item.id, "gustavo", sess) == 1  # not draft anymore
+    assert item.operator_signature == sig and len(sig) == 128
+    assert publisher._approve(item.id, "gustavo", sess, signature=sig) == 1  # not draft anymore
 
     actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor
     assert actor == "gustavo"
@@ -78,9 +108,7 @@ def test_approve_gate(cfg, sess):
 
 
 def test_publish_due_nostr_happy_path(cfg, sess):
-    item = _item()  # approved, scheduled None -> due now
-    sess.add(item)
-    sess.flush()
+    item = _approved_item(cfg, sess)
 
     relays_seen = []
 
@@ -99,25 +127,69 @@ def test_publish_due_nostr_happy_path(cfg, sess):
     assert item.status is ContentStatus.PUBLISHED
     assert item.post_ref and len(item.post_ref) == 64
     assert item.published_at is not None
-    assert "wss://relay.example.org" in relays_seen[0]  # noqa: E501
+    assert "wss://relay.example.org" in relays_seen[0]
 
 
 def test_future_scheduled_item_not_published(cfg, sess):
-    from datetime import UTC, datetime, timedelta
-
-    item = _item(scheduled=datetime.now(UTC) + timedelta(hours=2))
-    sess.add(item)
-    sess.flush()
-    config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org")
-    stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: [])
+    item = _approved_item(
+        cfg, sess, scheduled=datetime.now(UTC) + timedelta(hours=2)
+    )
+    config = cfg(
+        CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org"
+    )
+    stats = publisher._publish_due(
+        config, sess, nostr_sender=lambda r, e, timeout=10: []
+    )
     assert stats["published"] == 0
     assert item.status is ContentStatus.APPROVED
 
 
-def test_nostr_without_config_fails_clean(cfg, sess):
-    item = _item()
+def test_unsigned_sql_approved_item_not_published(cfg, sess):
+    """Round-2 required test: raw-SQL approved row (no signature) not published."""
+    config = cfg(
+        CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org"
+    )
+    item = _item()  # status=APPROVED, no signature
     sess.add(item)
     sess.flush()
+    stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: [])
+    assert stats["published"] == 0 and stats["tampered"] == 1
+    assert item.status is ContentStatus.DRAFT  # reverted
+    refused = sess.query(AuditEntry).filter_by(action="publish_refused").one()
+    assert refused.result == "error"
+
+
+def test_tampered_caption_refused_by_signature(cfg, sess):
+    """Round-2 required test: signature over other content refused at publish."""
+    item = _approved_item(cfg, sess)
+
+    item.caption = item.caption + " EDITED"  # content no longer matches sig
+    nostr_config = cfg(
+        CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org"
+    )
+    stats = publisher._publish_due(
+        nostr_config, sess, nostr_sender=lambda r, e, timeout=10: []
+    )
+    assert stats["published"] == 0 and stats["tampered"] == 1
+    assert item.status is ContentStatus.DRAFT
+
+
+def test_foreign_deployment_signature_refused(cfg, sess, signer):
+    """Round-2 required test: a valid signature for another deployment_id is
+    refused at approval time."""
+    item = _item(status=ContentStatus.DRAFT)
+    sess.add(item)
+    sess.flush()
+    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 item.status is ContentStatus.DRAFT
+
+
+def test_nostr_without_config_fails_clean(cfg, sess):
+    """Signed item but no nostr credentials -> clean failure, item stays."""
+    item = _approved_item(cfg, sess)
     config = cfg()  # no key, no relays
     stats = publisher._publish_due(config, sess)
     assert stats["failed"] == 1
@@ -126,14 +198,15 @@ def test_nostr_without_config_fails_clean(cfg, sess):
 
 
 def test_non_nostr_channels_skipped_terminal(cfg, sess):
-    item = _item(channel="facebook", theme="lightning-abroad")
-    sess.add(item)
-    sess.flush()
+    item = _approved_item(cfg, sess, channel="facebook", theme="lightning-abroad")
     stats = publisher._publish_due(cfg(), sess)
     assert stats["skipped"] == 1
     assert item.status is ContentStatus.SKIPPED  # terminal (audit fix)
-    skip = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").one()
-    assert skip.detail["channel"] == "facebook"
+
+    stats2 = publisher._publish_due(cfg(), sess)
+    assert stats2["skipped"] == 0  # never revisited
+    skips = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").count()
+    assert skips == 1
 
 
 def test_meta_adapters_disabled_by_default(cfg):
@@ -164,61 +237,27 @@ def test_unknown_brand_rejected():
         images.render_brand_image(brand="xx", kicker=None, lines=["A"], out_path="/tmp/x.jpg")
 
 
-# --- audit corrections ----------------------------------------------------------
-
-
-def test_publish_refuses_tampered_caption(cfg, sess):
-    """Medium finding: approved caption was not bound to content."""
-    item = _item(status=ContentStatus.DRAFT)
-    sess.add(item)
-    sess.flush()
-    assert publisher._approve(item.id, "gustavo", sess) == 0
-
-    item.caption = item.caption + " TAMPERED"
-    config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org")
-    stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: [])
-
-    assert stats["published"] == 0 and stats["tampered"] == 1
-    assert item.status is ContentStatus.APPROVED  # stays, needs re-approval
-    action = sess.query(AuditEntry).filter_by(action="publish_refused_hash_mismatch").one()
-    assert action.result == "error"
-
-
-def test_disabled_channel_is_terminal(cfg, sess):
-    """Medium finding: disabled channels wrote skip audits every 5 minutes."""
-    item = _item(channel="facebook", theme="lightning-abroad")
-    sess.add(item)
-    sess.flush()
-    config = cfg()
-
-    stats1 = publisher._publish_due(config, sess)
-    assert stats1["skipped"] == 1
-    assert item.status is ContentStatus.SKIPPED  # terminal
-
-    stats2 = publisher._publish_due(config, sess)
-    assert stats2["skipped"] == 0  # never revisited
-    skips = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").count()
-    assert skips == 1
+# --- round-2: attempts budget --------------------------------------------------
 
 
 def test_publish_attempts_bounded(cfg, sess, monkeypatch):
     """Medium finding: nostr failures retried without limit."""
     from cayceops.components.publisher import nostr as nostr_mod
 
-    item = _item()
-    sess.add(item)
-    sess.flush()
-
-    def always_fail(relays, event, timeout=10.0):
-        raise RuntimeError("relay down")
-
-    monkeypatch.setattr(nostr_mod, "publish", always_fail)
     config = cfg(
         CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY,
         CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org",
         CAYCE_OPS_MAX_ATTEMPTS="1",
     )
+    item = _approved_item(cfg, sess)
 
+    def always_fail(relays, event, timeout=10.0):
+        raise RuntimeError("relay down")
+
+    monkeypatch.setattr(nostr_mod, "publish", always_fail)
     publisher._publish_due(config, sess, nostr_sender=always_fail)
     assert item.attempts == 1
     assert item.status is ContentStatus.SKIPPED  # budget exhausted -> terminal
+
+
+import os  # noqa: E402  (used by image test)