From a542403e29e8268a6ebf941f0b7593aa6d3fc295 Mon Sep 17 00:00:00 2001 From: ABrandes <79119643+AMBRA7592@users.noreply.github.com> Date: Sat, 18 Jul 2026 06:49:56 +0200 Subject: [PATCH] Governance: require explicit off-menu decisions --- govern.py | 19 ++++++++++ soft_labels.py | 1 + test_claims.py | 100 ++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 1 deletion(-) diff --git a/govern.py b/govern.py index 28e0476..2514b57 100644 --- a/govern.py +++ b/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" diff --git a/soft_labels.py b/soft_labels.py index b0ff192..a9c1871 100644 --- a/soft_labels.py +++ b/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 diff --git a/test_claims.py b/test_claims.py index fa7d370..e51b449 100644 --- a/test_claims.py +++ b/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)