NK-WP-0056-T07: founder kill switch and execution findings in posture feedback

- cli: SIGTERM/SIGINT (kubectl delete job) sets the stop flag; the runner stops at the next
  step boundary and reverts. A stop before consume refuses and spends nothing.
- posture feedback v0.1 section 4a: --receipts reads bounded-executor events through the
  receipt validator: execution-receipt-missing, -not-completed, -reverted, -budget-pressure,
  and the aggregate execution-recurring-revert. Owners come from the authoritative
  tools/posture-feedback/execution-owners.yaml, never inferred.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 523928@bnt-lap001
Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350

bounded-executor/cli.py

@@ -17,6 +17,11 @@ The config (JSON, non-secret) names endpoints, token files and the pilot's targe
 Exit codes: 0 completed, 3 reverted, 4 refused before any effect, 5 stopped / unknown / revert failed / budget,
 6 consumed but not in custody. The result printed is value-free.
 
+Kill switch (NK-WP-0056-T07): SIGTERM or SIGINT, e.g. the founder's `kubectl -n sso delete job <job>`, sets the stop
+flag. Before consume the run refuses and spends nothing; after it, the runner finishes the step in flight, then
+stops and reverts what it changed (BEA stop_and_revert). The Job's terminationGracePeriodSeconds must cover a
+revert.
+
   python3 bounded-executor/cli.py --journey-only --config FILE [--result FILE]
 
   python3 bounded-executor/cli.py --request --admission FILE --config FILE [--result FILE]
@@ -46,7 +51,9 @@ import argparse
 import importlib.util
 import json
 import pathlib
+import signal
 import sys
+import threading
 
 HERE = pathlib.Path(__file__).resolve().parent
 
@@ -63,6 +70,14 @@ X = _load("executor")
 OPS = _load("operations")
 C = _load("clients")
 EXIT = {"completed": 0, "reverted": 3, "refused": 4}
+STOP = threading.Event()  # the kill switch; set by SIGTERM/SIGINT, read between steps
+
+
+def install_kill_switch() -> None:
+    def stop(_signum, _frame):
+        STOP.set()  # never raises into the step in flight: the runner stops at the next boundary and reverts
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
 
 
 def approval_carrier(cfg: dict, logins: list):
@@ -125,7 +140,7 @@ def build(cfg: dict, logins: list):
                   operations=operations, checks=checks, trust=lambda state: state in trusted,
                   audit=C.AuditCore(cfg["audit_core"]["url"], cfg["audit_core"]["token_file"],
                                     cfg["audit_core"].get("ca_file")),
-                  alert=alert)
+                  alert=alert, should_stop=STOP.is_set)
     config = X.Config(executor_identity=cfg["executor_identity"], runtime_principal=cfg["runtime_principal"],
                       tenant=cfg["tenant"], environment=cfg["environment"],
                       approver_subjects=dict(cfg["approver_subjects"]),
@@ -316,6 +331,7 @@ def main(argv=None) -> int:
         result, code = run_journey_only(cfg)
         _emit(result, args.result)
         return code
+    install_kill_switch()
     admission = X.bea.load(pathlib.Path(args.admission))
     logins: list = []
     try:

bounded-executor/executor.py

@@ -354,6 +354,8 @@ class Run:
             self.preflight()
         except Refused as refusal:
             return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False}
+        if self.d.should_stop():  # the founder's stop before the one irreversible act: nothing is spent
+            return {"outcome": "refused", "reason": "stopped_before_consume", "run_id": None, "effects": False}
         self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic()
         try:
             self.consume()

bounded-executor/tests/test_approval_request.py

@@ -336,5 +336,20 @@ class RequestPhase(Base):
         self.assertNotIn(SECRET, text)
 
 
+
+
+class KillSwitch(unittest.TestCase):
+    def test_sigterm_sets_the_stop_flag_the_runner_reads(self):
+        import os
+        import signal
+        before = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT)}
+        self.addCleanup(lambda: [signal.signal(sig, h) for sig, h in before.items()])
+        self.addCleanup(CLI.STOP.clear)
+        CLI.STOP.clear()
+        CLI.install_kill_switch()
+        os.kill(os.getpid(), signal.SIGTERM)
+        self.assertTrue(CLI.STOP.wait(2))
+
+
 if __name__ == "__main__":
     unittest.main()

bounded-executor/tests/test_executor.py

@@ -402,6 +402,12 @@ class FailureTests(Base):
         self.assertIn("restore-kv", self.ops())
         self.assertNotIn("deliver", self.ops())
 
+    def test_stop_before_consume_spends_nothing(self):
+        self.w.stop_after = 0
+        result = self.run_it()
+        self.assertEqual((result["outcome"], result["reason"]), ("refused", "stopped_before_consume"))
+        self.assertEqual((self.w.carrier.consumes, self.w.events, self.ops()), (0, [], []))
+
     def test_settle_time_is_waited_and_counted(self):
         self.doc = synthetic_admission(self.root, lambda s: s["checks"][3].update(settle_seconds=125))
         self.w = World(self.doc, self.clock)

canon/schemas/posture-feedback-report_v0.1.schema.json

@@ -112,7 +112,12 @@
             "evidence-expired",
             "evidence-due-soon",
             "implemented-not-evidenced",
-            "declared-gap"
+            "declared-gap",
+            "execution-receipt-missing",
+            "execution-not-completed",
+            "execution-reverted",
+            "execution-budget-pressure",
+            "execution-recurring-revert"
           ]
         },
         "severity": {

canon/standards/posture-feedback_v0.1.md

@@ -48,6 +48,8 @@ The evaluator routes only from authoritative declaration fields:
 - posture review, gaps, and implemented controls: `responsible_repo`;
 - evidence replacement: `evidence_freshness.<level>.responsible_repo`;
 - security-zone review: `zones.responsible_party`.
+- execution findings (§4a): `responsible_repo` for the receipt's `executing_workload` in
+  `tools/posture-feedback/execution-owners.yaml`.
 
 If the relevant field is absent, owner resolution is `unknown`. The evaluator
 must not infer ownership from the service name, repository path, Git remote,
@@ -75,6 +77,26 @@ Current adversarial levels are `E2`, `R4`, and `V2`–`V4`. This vocabulary come
 from Tenancy Posture §13. Mechanical evidence is evaluated for expiry only when
 its declaration explicitly supplies `valid_until`.
 
+## 4a. Execution findings (added 2026-11-04, NK-WP-0056-T07)
+
+With `--receipts`, the evaluator also reads bounded-executor events: `execution.run.started` and
+`execution.attribution.receipt`. Each event must pass the Execution Attribution Receipt v0.1 validator
+(`tools/execution-attribution-receipt/`). An invalid event, or a started/receipt pair that does not match (EAR-10),
+is a validation error, never a finding. Runs are joined by `run_id`. A finding's control is
+`admission:<id>#run:<run_id>`.
+
+| Finding | Trigger | Severity |
+| --- | --- | --- |
+| `execution-receipt-missing` | a started event with no receipt among the inputs | high |
+| `execution-not-completed` | outcome `unknown`, `revert_failed`, `stopped` or `budget` | high |
+| `execution-reverted` | outcome `reverted` | medium |
+| `execution-budget-pressure` | wall time used is at least 80 % of the admitted limit, or more than one attempt | low |
+| `execution-recurring-revert` | two or more reverted admissions for the same executor and consumer set | high |
+
+`execution-recurring-revert` is the aggregate review of recurring admissions. Its control is
+`consumers:<sorted consumers>` and its reason lists the admissions. The evaluator reads no wall clock here
+either. Receipts carry their own times, and the report's `as_of` stays explicit.
+
 ## 5. Proposal and safety boundary
 
 Every finding receives a stable id derived from its source declaration,

tools/posture-feedback/execution-owners.yaml

@@ -0,0 +1,6 @@
+# Authoritative owner routing for execution findings (posture-feedback v0.1 section 3, NK-WP-0056-T07).
+# executing_workload -> the repository responsible for the executor's behavior. The evaluator never infers
+# an owner; a workload not listed here is routed to `unknown`.
+executors:
+  - executing_workload: service:nk-bind-executor
+    responsible_repo: net-kingdom   # founder decision 6d1adce9: NetKingdom writes the executor

tools/posture-feedback/posture_feedback.py

@@ -24,6 +24,10 @@ ROOT = pathlib.Path(__file__).resolve().parents[2]
 TENANCY_SCHEMA = ROOT / "canon/schemas/tenancy-posture_v0.1.schema.json"
 REPORT_SCHEMA = ROOT / "canon/schemas/posture-feedback-report_v0.1.schema.json"
 TENANCY_VALIDATOR_PATH = ROOT / "tools/tenancy-posture/validate.py"
+RECEIPT_VALIDATOR_PATH = ROOT / "tools/execution-attribution-receipt/validate.py"
+EXECUTION_OWNERS = ROOT / "tools/posture-feedback/execution-owners.yaml"
+NOT_COMPLETED = {"unknown", "revert_failed", "stopped", "budget"}
+BUDGET_PRESSURE = 0.8  # share of the admitted wall time
 API_VERSION = "netkingdom.io/posture-feedback/v0.1"
 KIND = "PostureFeedbackReport"
 ADVERSARIAL_LEVELS = {"E2", "R4", "V2", "V3", "V4"}
@@ -35,6 +39,10 @@ SPEC = importlib.util.spec_from_file_location(
 assert SPEC and SPEC.loader
 TENANCY_VALIDATOR = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(TENANCY_VALIDATOR)
+_RSPEC = importlib.util.spec_from_file_location("receipt_validator_for_feedback", RECEIPT_VALIDATOR_PATH)
+assert _RSPEC and _RSPEC.loader
+RECEIPT_VALIDATOR = importlib.util.module_from_spec(_RSPEC)
+_RSPEC.loader.exec_module(RECEIPT_VALIDATOR)
 
 
 def parse_timestamp(value: str) -> dt.datetime:
@@ -321,6 +329,95 @@ def evaluate_entry(
     return results
 
 
+def execution_owners(path: pathlib.Path = EXECUTION_OWNERS) -> dict[str, str]:
+    """executing_workload -> responsible_repo, from the authoritative declaration (§3). Nothing is inferred."""
+    document = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+    return {e["executing_workload"]: e["responsible_repo"] for e in document.get("executors", [])}
+
+
+def execution_findings(paths: list[pathlib.Path], owners: dict[str, str]) -> tuple[list[dict], list[str]]:
+    """Proposal-only findings from bounded-executor events (§4a). Each file holds one event or a list of events.
+    Invalid events are validation errors, never findings. Runs are joined by run_id."""
+    errors: list[str] = []
+    runs: dict[str, dict[str, Any]] = {}
+    invalid: set[str] = set()  # a run with an invalid event is reported once, as a validation error
+    for path in paths:
+        source = source_identifier(path)
+        try:
+            loaded = json.loads(path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            errors.append(f"{path}: {exc}")
+            continue
+        for event in loaded if isinstance(loaded, list) else [loaded]:
+            report = RECEIPT_VALIDATOR.evaluate(event)
+            if not report["valid"]:
+                errors.extend(f"{path}: {f['rule']} at {f['at']}" for f in report["findings"])
+                if isinstance(event, dict) and isinstance(event.get("correlation_id"), str):
+                    invalid.add(event["correlation_id"])
+                continue
+            runs.setdefault(event["correlation_id"], {"source": source})[report["kind"]] = event
+
+    found: list[dict[str, Any]] = []
+    reverted_by_scope: dict[tuple[str, str], list[str]] = {}
+    for run_id, run in sorted(runs.items()):
+        if run_id in invalid:
+            continue
+        if "started" in run and "receipt" in run:
+            pair = RECEIPT_VALIDATOR.evaluate_pair(run["started"], run["receipt"])
+            if not pair["valid"]:
+                errors.extend(f"run {run_id}: {f['rule']} at {f['at']}" for f in pair["findings"])
+                continue
+        event = run.get("receipt") or run["started"]
+        data = event["data"]
+        service = data["executing_workload"]
+        owner = owners.get(service, "unknown")
+        subject = event["subject"]
+        control = f"{subject}#run:{run_id}"
+
+        def add(finding_class: str, severity: str, reason: str, action: str, state: str = "not-applicable"):
+            found.append(finding(source=run["source"], service=service, finding_class=finding_class,
+                                 severity=severity, control=control, owner=owner, evidence_state=state,
+                                 reason=reason, recommended_action=action))
+
+        if "receipt" not in run:
+            add("execution-receipt-missing", "high",
+                f"Run {run_id} for {subject} started (approval consumed) but no receipt is among the inputs.",
+                "Check audit-core's opening_without_closing finding and the Job's state; establish what the run did "
+                "before any further admission for this scope.", "unknown")
+            continue
+        consumers = sorted({s["consumer"] for s in data["steps"]})
+        scope = ",".join(consumers) or "none"
+        outcome = data["outcome"]
+        if outcome in NOT_COMPLETED:
+            add("execution-not-completed", "high",
+                f"Run {run_id} for {subject} ended {outcome}; the target state is not established by the executor.",
+                "Inspect the receipt's steps and the target; restore a known state under a new admission or by hand.",
+                "unknown")
+        elif outcome == "reverted":
+            add("execution-reverted", "medium",
+                f"Run {run_id} for {subject} reverted after a failed check or definite failure.",
+                "Review the failing step and check before re-admitting the same plan.")
+            reverted_by_scope.setdefault((service, scope), []).append(event["subject"])
+        budget = data["budget"]
+        if budget["used"]["wall_time_seconds"] >= BUDGET_PRESSURE * budget["limit"]["wall_time_seconds"] \
+                or data["attempts"] > 1:
+            add("execution-budget-pressure", "low",
+                f"Run {run_id} for {subject} used {budget['used']['wall_time_seconds']:g}s of "
+                f"{budget['limit']['wall_time_seconds']}s and {data['attempts']} of {budget['limit']['attempts']} "
+                "attempt(s).",
+                "Review whether the admitted budget fits the plan before the next admission of this kind.")
+    for (service, scope), controls in sorted(reverted_by_scope.items()):
+        if len(controls) >= 2:
+            found.append(finding(
+                source="execution-aggregate", service=service, finding_class="execution-recurring-revert",
+                severity="high", control=f"consumers:{scope}", owner=owners.get(service, "unknown"),
+                evidence_state="not-applicable",
+                reason=f"{len(controls)} admissions for consumers {scope} reverted: {', '.join(sorted(controls))}.",
+                recommended_action="Review the recurring admission in aggregate: the plan, its checks or the target "
+                                   "drifted; stop admitting it until reviewed."))
+    return found, errors
+
+
 def report_digest(report_without_digest: dict[str, Any]) -> str:
     encoded = json.dumps(
         report_without_digest,
@@ -336,6 +433,8 @@ def build_report(
     as_of: dt.datetime,
     horizon_days: int,
     fail_on: str,
+    receipts: list[pathlib.Path] | None = None,
+    owners: dict[str, str] | None = None,
 ) -> dict[str, Any]:
     tenancy_schema = json.loads(TENANCY_SCHEMA.read_text(encoding="utf-8"))
     validation_errors: list[str] = []
@@ -354,6 +453,10 @@ def build_report(
             continue
         for entry in service_entries(document):
             findings.extend(evaluate_entry(source, entry, as_of, horizon_days))
+    if receipts:
+        found, errors = execution_findings(receipts, execution_owners() if owners is None else owners)
+        findings.extend(found)
+        validation_errors.extend(errors)
 
     findings.sort(
         key=lambda item: (
@@ -408,7 +511,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         description="Emit deterministic, proposal-only NetKingdom posture feedback."
     )
-    parser.add_argument("declarations", nargs="+", type=pathlib.Path)
+    parser.add_argument("declarations", nargs="*", type=pathlib.Path)
+    parser.add_argument("--receipts", nargs="+", type=pathlib.Path, default=[],
+                        help="bounded-executor started/receipt events (JSON), section 4a")
     parser.add_argument("--as-of", required=True, help="RFC 3339 timestamp with explicit offset")
     parser.add_argument("--horizon-days", type=int, default=30)
     parser.add_argument("--fail-on", choices=("none", "low", "medium", "high"), default="high")
@@ -424,11 +529,14 @@ def main(argv: list[str] | None = None) -> int:
     except ValueError as exc:
         raise SystemExit(f"invalid --as-of: {exc}") from exc
 
+    if not args.declarations and not args.receipts:
+        raise SystemExit("give declarations, --receipts, or both")
     report = build_report(
         args.declarations,
         as_of=as_of,
         horizon_days=args.horizon_days,
         fail_on=args.fail_on,
+        receipts=args.receipts,
     )
     print(json.dumps(report, indent=2, sort_keys=True))
     if report["validation_errors"]:

tools/posture-feedback/tests/test_execution_feedback.py

@@ -0,0 +1,126 @@
+"""Execution findings from bounded-executor receipts (posture-feedback v0.1 section 4a, NK-WP-0056-T07)."""
+from __future__ import annotations
+
+import copy
+import datetime as dt
+import importlib.util
+import json
+import pathlib
+import sys
+import tempfile
+import unittest
+import uuid
+
+HERE = pathlib.Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+FIXTURES = ROOT / "tools/execution-attribution-receipt/fixtures/valid"
+spec = importlib.util.spec_from_file_location("posture_feedback_exec", HERE.parent / "posture_feedback.py")
+PF = importlib.util.module_from_spec(spec)
+sys.modules["posture_feedback_exec"] = PF
+spec.loader.exec_module(PF)
+AS_OF = dt.datetime(2026, 11, 4, 12, 0, tzinfo=dt.UTC)
+
+
+def events(name):
+    return [json.loads((FIXTURES / f"{name}-{k}.json").read_text()) for k in ("started", "receipt")]
+
+
+def renamed(evs, admission):
+    """The same run shape as another run of another admission (fresh run id)."""
+    rid = str(uuid.uuid4())
+    out = []
+    for e in copy.deepcopy(evs):
+        kind = "started" if e["type"] == "execution.run.started" else "receipt"
+        e.update(id=f"run:{rid}:{kind}", correlation_id=rid, subject=f"admission:{admission}")
+        e["data"]["run_id"] = rid
+        if kind == "started":
+            e["data"]["admission"] = admission
+        else:
+            e["data"]["target"] = f"admission:{admission}"
+            e["data"]["decision_refs"][0]["ref"] = admission
+        out.append(e)
+    return out
+
+
+class ExecutionFeedback(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.dir = pathlib.Path(tmp.name)
+
+    def report(self, *runs, owners=None, fail_on="none"):
+        paths = []
+        for i, evs in enumerate(runs):
+            path = self.dir / f"run-{i}.json"
+            path.write_text(json.dumps(evs))
+            paths.append(path)
+        return PF.build_report([], as_of=AS_OF, horizon_days=14, fail_on=fail_on, receipts=paths, owners=owners)
+
+    def classes(self, report):
+        return sorted((f["class"], f["severity"]) for f in report["findings"])
+
+    def test_a_clean_run_raises_nothing(self):
+        report = self.report(events("completed"))
+        self.assertEqual((report["findings"], report["validation_errors"], report["ok"]), ([], [], True))
+
+    def test_reverted_and_retry_findings_are_routed_to_the_declared_owner(self):
+        report = self.report(events("reverted"), events("completed-on-retry"))
+        self.assertEqual(self.classes(report), [("execution-budget-pressure", "low"), ("execution-reverted", "medium")])
+        self.assertEqual({f["owner"] for f in report["findings"]}, {"net-kingdom"})
+        self.assertEqual({f["service"] for f in report["findings"]}, {"service:nk-bind-executor"})
+        self.assertEqual(report["automation"]["mode"], "proposal-only")
+
+    def test_an_unmapped_executor_is_unknown_never_inferred(self):
+        report = self.report(events("reverted"), owners={})
+        self.assertEqual({f["owner"] for f in report["findings"]}, {"unknown"})
+        self.assertEqual(report["summary"]["unknown_owner"], 1)
+
+    def test_not_completed_is_high(self):
+        evs = events("reverted")
+        evs[1]["data"].update(outcome="unknown", reverted=True)
+        report = self.report(evs, fail_on="high")
+        self.assertIn(("execution-not-completed", "high"), self.classes(report))
+        self.assertFalse(report["ok"])
+
+    def test_a_start_without_receipt_is_high(self):
+        report = self.report(events("completed")[:1])
+        self.assertEqual(self.classes(report), [("execution-receipt-missing", "high")])
+
+    def test_recurring_reverts_are_reviewed_in_aggregate(self):
+        base = events("reverted")
+        report = self.report(base, renamed(base, "BEA-2026-0002"))
+        aggregate = [f for f in report["findings"] if f["class"] == "execution-recurring-revert"]
+        self.assertEqual(len(aggregate), 1)
+        self.assertEqual((aggregate[0]["severity"], aggregate[0]["control"]), ("high", "consumers:authelia"))
+        self.assertIn("BEA-2026-0001", aggregate[0]["reason"])
+        self.assertIn("BEA-2026-0002", aggregate[0]["reason"])
+
+    def test_invalid_events_are_validation_errors_not_findings(self):
+        evs = events("completed")
+        evs[1]["data"]["bind_password"] = "synthetic-value"
+        report = self.report(evs)
+        self.assertTrue(report["validation_errors"])
+        self.assertEqual(report["findings"], [])
+        self.assertNotIn("synthetic-value", json.dumps(report))
+
+    def test_a_mismatched_pair_is_a_validation_error(self):
+        started, receipt = events("completed")
+        started["data"]["approval"] = "another-approval"
+        report = self.report([started, receipt])
+        self.assertTrue(any("EAR-10" in e for e in report["validation_errors"]))
+
+    def test_deterministic(self):
+        first = self.report(events("reverted"), events("completed-on-retry"))
+        second = self.report(events("reverted"), events("completed-on-retry"))
+        self.assertEqual(first["report_digest"], second["report_digest"])
+
+    def test_cli_receipts_only(self):
+        path = self.dir / "r.json"
+        path.write_text(json.dumps(events("reverted")))
+        self.assertEqual(PF.main(["--as-of", "2026-11-04T12:00:00Z", "--fail-on", "medium", "--receipts", str(path)]), 1)
+        with self.assertRaises(SystemExit):
+            PF.main(["--as-of", "2026-11-04T12:00:00Z"])
+
+
+if __name__ == "__main__":
+    unittest.main()