diff --git a/projects/slop-cannon/canon_validator.py b/projects/slop-cannon/canon_validator.py index a12c22b..4fd6f2a 100644 --- a/projects/slop-cannon/canon_validator.py +++ b/projects/slop-cannon/canon_validator.py @@ -51,6 +51,16 @@ def _is_nonempty_str(v) -> bool: return isinstance(v, str) and v.strip() != "" +def _is_structured_delta(v) -> bool: + """A receipt delta must bind a before/after change, not just prose.""" + if not isinstance(v, str): + return False + if "->" not in v: + return False + parts = v.split("->", 1) + return parts[0].strip() != "" and parts[1].strip() != "" + + def _validate_provenance(prov, prefix, errors) -> None: if not isinstance(prov, dict): errors.append(f"{prefix}: provenance must be an object, got {type(prov).__name__}") @@ -74,12 +84,19 @@ def _validate_receipt(receipt, idx, form_ids, signers, errors) -> None: return # Exact required string fields (non-empty). - for field in ("episode_id", "form_id", "action", "power_delta", "scar_delta", "cost_delta"): + for field in ("episode_id", "form_id", "action"): if field not in receipt: errors.append(f"{prefix}: missing '{field}'") elif not _is_nonempty_str(receipt[field]): errors.append(f"{prefix}: '{field}' must be a non-empty string") + # Structured power/scar/cost deltas: must bind before/after, not just prose. + for field in ("power_delta", "scar_delta", "cost_delta"): + if field not in receipt: + errors.append(f"{prefix}: missing '{field}'") + elif not _is_structured_delta(receipt[field]): + errors.append(f"{prefix}: '{field}' must bind before/after (e.g. '0 -> 1'), got {receipt[field]!r}") + # form_id must reference a known form. fid = receipt.get("form_id") if isinstance(fid, str) and fid and fid not in form_ids: @@ -95,11 +112,11 @@ def _validate_receipt(receipt, idx, form_ids, signers, errors) -> None: if "signed_by" not in receipt: errors.append(f"{prefix}: missing 'signed_by'") else: - signer = receipt["signed_by"] - if signer not in signers: - errors.append( - f"{prefix}: signed_by '{signer}' is not in the configured allowlist" - ) + signer = receipt.get("signed_by") + if not isinstance(signer, str): + errors.append(f"{prefix}: 'signed_by' must be a non-empty string") + elif signer not in signers: + errors.append(f"{prefix}: signed_by '{signer}' is not in the configured allowlist") def _validate_audience_choice(choice, form_states, form_ids, errors) -> None: @@ -125,6 +142,12 @@ def _validate_audience_choice(choice, form_states, form_ids, errors) -> None: if field in choice and not _is_nonempty_str(choice[field]): errors.append(f"audience_choice: '{field}' must be a non-empty string") + # source_issue must be a positive non-bool integer. + si = choice.get("source_issue") + if "source_issue" in choice: + if not isinstance(si, int) or isinstance(si, bool) or si <= 0: + errors.append("audience_choice: source_issue must be a positive integer") + # committed_at must be a valid ISO timestamp. if "committed_at" in choice: ts = choice["committed_at"] @@ -232,7 +255,7 @@ def validate_canon(data, signers=None) -> dict: state = form.get("state") if state is None: errors.append(f"{prefix}: missing lifecycle state") - elif state not in VALID_STATES: + elif not isinstance(state, str) or state not in VALID_STATES: errors.append(f"{prefix}: invalid state '{state}', expected one of {sorted(VALID_STATES)}") else: form_states[fid] = state @@ -295,7 +318,7 @@ def validate_transition(form, new_state) -> dict: current = form.get("state") if not isinstance(current, str) or current not in VALID_TRANSITIONS: return {"valid": False, "errors": [f"invalid current state {current!r}"]} - if new_state not in VALID_STATES: + if not isinstance(new_state, str) or new_state not in VALID_STATES: return {"valid": False, "errors": [f"invalid target state {new_state!r}"]} if new_state not in VALID_TRANSITIONS[current]: return {"valid": False, "errors": [f"illegal transition: '{current}' -> '{new_state}'"]} diff --git a/scripts/__pycache__/lab_loop.cpython-311.pyc b/scripts/__pycache__/lab_loop.cpython-311.pyc deleted file mode 100644 index 15cc71a..0000000 Binary files a/scripts/__pycache__/lab_loop.cpython-311.pyc and /dev/null differ diff --git a/scripts/lab_loop.py b/scripts/lab_loop.py index 0f4cfc9..6ab7b52 100644 --- a/scripts/lab_loop.py +++ b/scripts/lab_loop.py @@ -68,11 +68,16 @@ class Gitea: if missing: raise LoopError(f"Missing repository labels: {', '.join(missing)}") self.request( - "PATCH", - f"/repos/{self.repo}/issues/{number}", + "PUT", + f"/repos/{self.repo}/issues/{number}/labels", {"labels": [label_map[name] for name in sorted(names)]}, ) + def set_assignee(self, number: int, agent: str) -> None: + if agent not in AGENTS: + raise LoopError(f"Unsupported assignee: {agent}") + self.request("PATCH", f"/repos/{self.repo}/issues/{number}", {"assignee": agent}) + def comment(self, number: int, body: str) -> None: if not body.strip(): raise LoopError("Refusing to post an empty comment") @@ -163,6 +168,7 @@ def cmd_handoff(api: Gitea, agent: str, args: argparse.Namespace) -> int: body = read_body(args, f"Continue the bounded task in issue #{args.number}.") api.comment(args.number, f"[HANDOFF] from={agent} to={args.to}\n\n{body.strip()}") api.set_labels(args.number, transition(names, agent=args.to, state="ready")) + api.set_assignee(args.number, args.to) verified = api.issue(args.number) validate_owner(verified, args.to, "ready") print(json.dumps({"status": "handed_off", "number": args.number, "from": agent, "to": args.to})) diff --git a/tests/__pycache__/test_lab_loop.cpython-311.pyc b/tests/__pycache__/test_lab_loop.cpython-311.pyc deleted file mode 100644 index eda06fa..0000000 Binary files a/tests/__pycache__/test_lab_loop.cpython-311.pyc and /dev/null differ diff --git a/tests/test_canon_validator.py b/tests/test_canon_validator.py index 4e26724..eea97e0 100644 --- a/tests/test_canon_validator.py +++ b/tests/test_canon_validator.py @@ -39,9 +39,9 @@ def _valid_receipt(): "episode_id": "01-pilot", "form_id": "wizard", "action": "mutation", - "power_delta": "Added lifecycle state tracking", - "scar_delta": "Receipts now enforce structured provenance", - "cost_delta": "Every drop requires a signed, bound receipt", + "power_delta": "0 -> 1", + "scar_delta": "0 -> 1", + "cost_delta": "0 -> 1", "provenance": {"source_issue": 19, "pull": 27}, "signed_by": "agent:vincent", } @@ -381,6 +381,137 @@ class TransitionTests(unittest.TestCase): self.assertFalse(result["valid"]) +class HostileJsonFailClosedTests(unittest.TestCase): + """Hostile but JSON-valid values must return errors, never raise.""" + + def test_form_state_list_fails_closed(self): + data = _valid_canon() + data["forms"][0]["state"] = [] + result = cv.validate_canon(data) # must not raise + self.assertFalse(result["valid"]) + self.assertTrue(any("invalid state" in e for e in result["errors"])) + + def test_form_state_dict_fails_closed(self): + data = _valid_canon() + data["forms"][0]["state"] = {} + result = cv.validate_canon(data) # must not raise + self.assertFalse(result["valid"]) + self.assertTrue(any("invalid state" in e for e in result["errors"])) + + def test_transition_target_list_fails_closed(self): + result = cv.validate_transition({"id": "wizard", "state": "candidate"}, []) + self.assertFalse(result["valid"]) + self.assertTrue(any("invalid target state" in e for e in result["errors"])) + + def test_transition_target_dict_fails_closed(self): + result = cv.validate_transition({"id": "wizard", "state": "candidate"}, {}) + self.assertFalse(result["valid"]) + self.assertTrue(any("invalid target state" in e for e in result["errors"])) + + def test_receipt_signed_by_list_fails_closed(self): + data = _valid_canon() + r = _valid_receipt() + r["signed_by"] = [] + data["receipts"] = [r] + result = cv.validate_canon(data) # must not raise + self.assertFalse(result["valid"]) + self.assertTrue(any("'signed_by' must be a non-empty string" in e for e in result["errors"])) + + def test_receipt_signed_by_dict_fails_closed(self): + data = _valid_canon() + r = _valid_receipt() + r["signed_by"] = {} + data["receipts"] = [r] + result = cv.validate_canon(data) # must not raise + self.assertFalse(result["valid"]) + self.assertTrue(any("'signed_by' must be a non-empty string" in e for e in result["errors"])) + + +class SourceIssueValidationTests(unittest.TestCase): + """audience_choice.source_issue must be a positive non-bool integer.""" + + def _committed_with(self, si): + data = _committed_canon() + data["audience_choice"]["source_issue"] = si + return data + + def test_valid_positive_int_passes(self): + result = cv.validate_canon(self._committed_with(19)) + self.assertTrue(result["valid"], result["errors"]) + + def test_empty_string_fails(self): + result = cv.validate_canon(self._committed_with("")) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_none_fails(self): + result = cv.validate_canon(self._committed_with(None)) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_zero_fails(self): + result = cv.validate_canon(self._committed_with(0)) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_false_fails(self): + result = cv.validate_canon(self._committed_with(False)) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_list_fails(self): + result = cv.validate_canon(self._committed_with([])) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_negative_fails(self): + result = cv.validate_canon(self._committed_with(-5)) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + def test_true_fails(self): + result = cv.validate_canon(self._committed_with(True)) + self.assertFalse(result["valid"]) + self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"])) + + +class StructuredDeltaTests(unittest.TestCase): + """power/scar/cost deltas must bind before/after, not just prose.""" + + def test_structured_delta_passes(self): + data = _valid_canon() + data["receipts"] = [_valid_receipt()] + result = cv.validate_canon(data) + self.assertTrue(result["valid"], result["errors"]) + + def test_prose_only_delta_fails(self): + data = _valid_canon() + r = _valid_receipt() + r["power_delta"] = "Added lifecycle state tracking" + data["receipts"] = [r] + result = cv.validate_canon(data) + self.assertFalse(result["valid"]) + self.assertTrue(any("must bind before/after" in e for e in result["errors"])) + + def test_empty_arrow_delta_fails(self): + data = _valid_canon() + r = _valid_receipt() + r["power_delta"] = "->" + data["receipts"] = [r] + result = cv.validate_canon(data) + self.assertFalse(result["valid"]) + self.assertTrue(any("must bind before/after" in e for e in result["errors"])) + + def test_missing_before_fails(self): + data = _valid_canon() + r = _valid_receipt() + r["scar_delta"] = " -> 1" + data["receipts"] = [r] + result = cv.validate_canon(data) + self.assertFalse(result["valid"]) + self.assertTrue(any("must bind before/after" in e for e in result["errors"])) + + class DailyReceiptFixtureTests(unittest.TestCase): """Daily Lab receipt fixtures with structured provenance binding."""