Spec A: guard off-menu governance decisions

govern.py

@@ -35,6 +35,11 @@ def _parse_args(argv):
     decide_parser.add_argument("--owner", required=True, help="named decision owner")
     decide_parser.add_argument("--decision", required=True, help="recorded decision")
     decide_parser.add_argument("--rationale", required=True, help="decision rationale")
+    decide_parser.add_argument(
+        "--allow-other",
+        action="store_true",
+        help="permit a decision outside the cell's observed labels (recorded verbatim)",
+    )
     decide_parser.add_argument(
         "--out",
         default=None,
@@ -204,9 +209,23 @@ def decide(args):
             )
         )
 
+    options = set(target.get("soft_label", {}))
+    off_menu = decision not in options
+    if off_menu and not args.allow_other:
+        _fail(
+            "decision {!r} is not one of the recorded options ({}); "
+            "pass --allow-other to record a deliberate off-menu decision".format(
+                decision, ", ".join(sorted(options)) or "none"
+            )
+        )
+
     target["decision_required_from"] = owner
     target["decision_recorded"] = decision
     target["decision_rationale"] = rationale
+    if off_menu:
+        target["decision_off_menu"] = True
+    else:
+        target.pop("decision_off_menu", None)
     target.pop("status", None)
     target["decided_at"] = utc_timestamp()
     target["status"] = "decided"

soft_labels.py

@@ -187,6 +187,7 @@ def main(argv=None):
             g["decision_recorded"] = p.get("decision_recorded")
             g["decision_rationale"] = p.get("decision_rationale")
             if "decided_at" in p: g["decided_at"] = p["decided_at"]
+            if "decision_off_menu" in p: g["decision_off_menu"] = p["decision_off_menu"]
         g["status"] = "decided" if g["decision_recorded"] is not None else "pending"
         merged.append(g)
     for key, p in prior.items():                         # don't drop a decided record just

test_claims.py

@@ -1049,11 +1049,20 @@ def _prepare_queue(self, out_dir):
             )
 
     def test_default_governance_bytes_are_pinned(self):
-        digest = hashlib.sha256(_pipeline()["governance_bytes"]).hexdigest()
+        pipeline = _pipeline()
+        digest = hashlib.sha256(pipeline["governance_bytes"]).hexdigest()
         self.assertEqual(
             digest,
             "91c1876d468d02694ec302158668e37d7482914be70b5ab6e8611c54cd3a8e2f",
         )
+        self.assertEqual(len(pipeline["governance"]), 2)
+        self.assertTrue(
+            all(
+                record.get("status") == "pending"
+                and "decision_off_menu" not in record
+                for record in pipeline["governance"]
+            )
+        )
 
     def test_decide_reaches_resolution_and_survives_exporter_rerun(self):
         with tempfile.TemporaryDirectory(prefix="groundless-govern-") as temp:
@@ -1110,6 +1119,7 @@ def test_decide_reaches_resolution_and_survives_exporter_rerun(self):
             )
             self.assertEqual(decision["status"], "decided")
             self.assertTrue(decision["decided_at"].endswith("Z"))
+            self.assertNotIn("decision_off_menu", decision)
 
             subprocess.run(
                 [sys.executable, str(ROOT / "resolution.py"), "--out", str(out_dir)],
@@ -1162,6 +1172,94 @@ def test_decide_reaches_resolution_and_survives_exporter_rerun(self):
             self.assertEqual(governance_path.read_bytes(), before_overwrite)
             self.assertEqual(list(out_dir.glob(".governance.*.tmp")), [])
 
+    def test_off_menu_decision_requires_explicit_escape_and_persists(self):
+        with tempfile.TemporaryDirectory(prefix="groundless-govern-off-menu-") as temp:
+            out_dir = Path(temp)
+            self._prepare_queue(out_dir)
+            governance_path = out_dir / "governance.jsonl"
+            original = governance_path.read_bytes()
+            command = [
+                sys.executable,
+                str(ROOT / "govern.py"),
+                "decide",
+                "--item",
+                "img2",
+                "--question",
+                "explicit",
+                "--owner",
+                "Safety policy owner",
+                "--decision",
+                "saef",
+                "--rationale",
+                "Apply a documented exception outside the observed labels.",
+                "--out",
+                str(out_dir),
+            ]
+
+            rejected = subprocess.run(
+                command,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                universal_newlines=True,
+            )
+            self.assertNotEqual(rejected.returncode, 0)
+            self.assertIn("explicit / flag", rejected.stderr)
+            self.assertIn("safe", rejected.stderr)
+            self.assertIn("--allow-other", rejected.stderr)
+            self.assertEqual(governance_path.read_bytes(), original)
+
+            subprocess.run(
+                command + ["--allow-other"],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                universal_newlines=True,
+            )
+            decision = next(
+                record
+                for record in _load_jsonl(governance_path)
+                if record["item_id"] == "img2" and record["question"] == "explicit"
+            )
+            self.assertEqual(decision["decision_recorded"], "saef")
+            self.assertIs(decision["decision_off_menu"], True)
+
+            subprocess.run(
+                [sys.executable, str(ROOT / "resolution.py"), "--out", str(out_dir)],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                universal_newlines=True,
+            )
+            resolution_record = next(
+                record
+                for record in _load_jsonl(out_dir / "resolution_records.jsonl")
+                if record["item"] == "img2" and record["question"] == "explicit"
+            )
+            self.assertEqual(resolution_record["disposition"]["outcome"], "decided:saef")
+            self.assertEqual(
+                resolution_record["authority"]["owner"], "Safety policy owner"
+            )
+            self.assertEqual(
+                resolution_record["authority"]["decided_by"], "named_owner"
+            )
+
+            subprocess.run(
+                [sys.executable, str(ROOT / "soft_labels.py"), "--out", str(out_dir)],
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                universal_newlines=True,
+            )
+            persisted = next(
+                record
+                for record in _load_jsonl(governance_path)
+                if record["item_id"] == "img2" and record["question"] == "explicit"
+            )
+            self.assertIs(persisted["decision_off_menu"], True)
+            self.assertEqual(persisted["decision_recorded"], "saef")
+            self.assertEqual(persisted["decided_at"], decision["decided_at"])
+            self.assertEqual(list(out_dir.glob(".governance.*.tmp")), [])
+
     def test_invalid_decisions_and_corrupt_decided_state_are_rejected(self):
         with tempfile.TemporaryDirectory(prefix="groundless-govern-invalid-") as temp:
             out_dir = Path(temp)