Route the sender lane's cluster writes through AUDIT_KUBECONFIG too.

Round-10 review (railiance-platform) found that ContainedOperator inherited
an ssh-only kubectl, so an E2 would write OpenBao and then fail its sync,
restart and rollback. Also: a bad AUDIT_KUBECONFIG now leaves window-error
evidence, and the receipt script validates bodies and lets a later
request-changes withdraw an approval.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 236506@bnt-lap001
Assistant-Session: 09087018-91e1-48c5-ab2f-acfbb62c92e4

docs/rpf-wp-0015-attended-windows.md

@@ -61,3 +61,11 @@ Cluster reads and the port-forward then run locally against that file; host
 reads such as uptime and boot id stay on `ssh railiance01`. Without the
 variable the scripts keep the earlier ssh path, which now fails at
 `read node state`.
+
+This includes the sender lane's projection and cleanup writes (ExternalSecret
+sync and audit-core restart). Set the variable on the one command only, never
+in the shell profile:
+
+```bash
+AUDIT_KUBECONFIG=$HOME/.kube/config-railiance01 make audit-t02-attended-window AUDIT_T02_MODE=e2 ...
+```

scripts/audit_core_synthetic_sender_lane.py

@@ -31,6 +31,7 @@ import os
 import re
 import secrets
 import stat
+import subprocess
 import sys
 import time
 from datetime import UTC, datetime, timedelta
@@ -280,6 +281,17 @@ class ContainedOperator(CREDS.Operator):
     def __init__(self, remote: str) -> None:  # noqa: D107 - deliberately no super().__init__
         self.remote = remote
         self.bao_env = LEASE.contained_bao_env()
+        # Same rule as PREFLIGHT.Remote: with AUDIT_KUBECONFIG set, cluster calls
+        # (including the projection/cleanup sync and restart writes) run locally
+        # against the attended admin kubeconfig; unset keeps ssh <remote> kubectl.
+        self.kubeconfig = PREFLIGHT.attended_kubeconfig()
+
+    def kubectl(self, args: list[str], *, label: str, input_text: str | None = None,
+                allow_missing: bool = False) -> subprocess.CompletedProcess[str]:
+        if self.kubeconfig is None:
+            return super().kubectl(args, label=label, input_text=input_text, allow_missing=allow_missing)
+        return CREDS.safe_run(["kubectl", "--kubeconfig", self.kubeconfig, *args], label=label,
+                              input_text=input_text, allow_missing=allow_missing)
 
 
 def contained_operator(args: argparse.Namespace) -> Any:

scripts/rpf-wp-0015-e2-receipt.py

@@ -7,12 +7,13 @@ The window is 15 minutes and starts now, or N minutes from now. Writing the
 receipt and running the window is the founder GO for that window; D5
 (2fd5849c) authorizes the exercise class only.
 
-The owner acknowledgements are the owners' latest State Hub `approve` receipts
+The owner acknowledgements are the owners' State Hub receipts for the
-for the *current* review contract
+*current* review contract
 (interfaces/RAILIANCE-WP-0024-owner-reviews.json). They are found by subject
-`WP0024-OWNER-RECEIPT/v1/<owner>/approve/<contract digest>`. The script
+`WP0024-OWNER-RECEIPT/v1/<owner>/<decision>/<contract digest>`, each body
-refuses when any pinned artifact differs from the contract, or when a
+validated by the review module. An owner's latest receipt must be `approve`.
-required owner has no approval for this digest. The receipt holds no
+The script refuses when any pinned artifact differs from the contract, or when
+a required owner has no current approval for this digest. The receipt holds no
 credential values, and the script never overwrites an existing file.
 """
 import importlib.util
@@ -42,12 +43,17 @@ def approvals(review, contract, api_base):
         messages = json.load(response)
     found = {}
     for owner in OWNERS:
-        subject = f"{review.RECEIPT_PREFIX}/v1/{owner}/approve/{digest}"
+        # Validate each body with the review module's own parser (subject, sender,
-        hits = sorted((m for m in messages if m.get("subject") == subject and m.get("from_agent") == owner),
+        # schema). The latest receipt for this digest decides: a request-changes
-                      key=lambda m: m["created_at"])
+        # after an approval withdraws it.
-        if not hits:
+        receipts = []
-            sys.exit(f"no {owner} approval for contract {digest[:8]}; run the owner round first")
+        for message in messages:
-        found[owner] = {"acknowledged": True, "message_id": hits[-1]["id"]}
+            receipt = review.parse_receipt_message(message)
+            if receipt and receipt["owner"] == owner and receipt["contract_digest"] == digest:
+                receipts.append((receipt["created_at"], receipt["decision"], message["id"]))
+        if not receipts or max(receipts)[1] != "approve":
+            sys.exit(f"no current {owner} approval for contract {digest[:8]}; run the owner round first")
+        found[owner] = {"acknowledged": True, "message_id": max(receipts)[2]}
     return digest, found
 
 

scripts/t02-attended-window.py

@@ -271,7 +271,8 @@ def main(argv: list[str] | None = None) -> int:
     # so this script never prints; its outcome lives only in the evidence dir.
     try:
         window(args)
-    except (OSError, KeyError, WindowError, LANE.LaneError, LANE.LEASE.ProcedureError) as exc:
+    except (OSError, KeyError, WindowError, LANE.LaneError, LANE.LEASE.ProcedureError,
+            LANE.PREFLIGHT.PreflightError) as exc:
         try:
             LANE.emit_result(
                 {"status": "failed", "error": str(exc), "secret_values_observed": False},

tests/test_audit_attended_kubeconfig.py

@@ -70,3 +70,40 @@ def test_bad_path_is_refused(monkeypatch, value):
     monkeypatch.setenv("AUDIT_KUBECONFIG", value)
     with pytest.raises(PRE.PreflightError):
         PRE.Remote("railiance01")
+
+
+LANE = load("lane_kc", "scripts/audit_core_synthetic_sender_lane.py")
+
+
+def operator(monkeypatch):
+    monkeypatch.setattr(LANE.LEASE, "contained_bao_env", lambda: {"BAO_ADDR": "http://127.0.0.1:18200"})
+    return LANE.ContainedOperator("railiance01")
+
+
+def test_lane_operator_writes_use_the_attended_kubeconfig(monkeypatch, tmp_path):
+    kc = tmp_path / "config"; kc.write_text("not read")
+    monkeypatch.setenv("AUDIT_KUBECONFIG", str(kc))
+    op = operator(monkeypatch)
+    with patch.object(LANE.CREDS.subprocess, "run", return_value=Done()) as run:
+        op.kubectl(["-n", "audit-core", "rollout", "restart", "deploy/audit-core"], label="restart audit-core")
+    assert run.call_args.args[0] == ["kubectl", "--kubeconfig", str(kc), "-n", "audit-core",
+                                     "rollout", "restart", "deploy/audit-core"]
+
+
+def test_lane_operator_unset_keeps_ssh(monkeypatch):
+    monkeypatch.delenv("AUDIT_KUBECONFIG", raising=False)
+    op = operator(monkeypatch)
+    with patch.object(LANE.CREDS.subprocess, "run", return_value=Done()) as run:
+        op.kubectl(["get", "pods"], label="read pods")
+    assert run.call_args.args[0][:3] == ["ssh", "-o", "BatchMode=yes"]
+
+
+def test_window_records_bad_kubeconfig_as_window_error(monkeypatch, tmp_path):
+    monkeypatch.setenv("AUDIT_KUBECONFIG", "relative/config")
+    evidence = tmp_path / "evidence"; evidence.mkdir(mode=0o700)
+    monkeypatch.setattr(WIN, "window", lambda args: WIN.PortForward("railiance01", 18080))
+    code = WIN.main(["--approval", str(tmp_path / "a.json"), "--load-driver", str(tmp_path / "d"),
+                     "--bearer-file", str(tmp_path / "b"), "--evidence-dir", str(evidence),
+                     "--confirm", "x"])
+    assert code == 1
+    assert [p.name.startswith("window-error-") for p in evidence.iterdir()] == [True]

tests/test_rpf_wp_0015_e2_receipt.py

@@ -47,3 +47,46 @@ def test_refuses_to_overwrite(tmp_path):
     result = subprocess.run([sys.executable, str(ROOT / "scripts/rpf-wp-0015-e2-receipt.py"), str(out)],
                             capture_output=True, text=True)
     assert result.returncode != 0 and out.read_text() == "{}"
+
+
+def test_later_request_changes_withdraws_an_approval():
+    gen = generator()
+    review = gen.review_module()
+    contract = review.load_contract()
+    digest = review.contract_digest(contract)
+
+    def message(decision, created, mid, sender="audit-core", owner="audit-core"):
+        body = review.build_receipt(contract, owner, decision, "claude/x",
+                                    verification={"passed": True, "checks": []},
+                                    note="needs a fix in file.py:1" if decision == "request-changes" else None)
+        body["created_at"] = created
+        return {"id": mid, "from_agent": sender, "subject": review.receipt_subject(body), "body": json.dumps(body)}
+
+    class Resp:
+        def __init__(self, data): self.data = data
+        def __enter__(self): return self
+        def __exit__(self, *a): return False
+        def read(self): return json.dumps(self.data).encode()
+
+    def fetch(msgs):
+        import unittest.mock as um
+        with um.patch.object(gen.urllib.request, "urlopen", return_value=Resp(msgs)):
+            with um.patch.object(gen.json, "load", lambda r: msgs):
+                return gen.approvals(review, contract, "http://hub")
+
+    ok = [message("approve", "2026-11-03T10:00:00Z", f"{o}-1", o, o) for o in gen.OWNERS]
+    assert fetch(ok)[0] == digest
+    withdrawn = ok + [message("request-changes", "2026-11-03T11:00:00Z", "ac-2")]
+    try:
+        fetch(withdrawn)
+    except SystemExit as exc:
+        assert "audit-core" in str(exc)
+    else:
+        raise AssertionError("withdrawn approval was accepted")
+    forged = [message("approve", "2026-11-03T10:00:00Z", "f", "someone-else", "audit-core")] + ok[1:]
+    try:
+        fetch(forged)
+    except SystemExit:
+        pass
+    else:
+        raise AssertionError("approval from the wrong sender was accepted")