feat: add Track A authoring workflow

dev/optimizer-evals/test_authoring_budget.py

@@ -0,0 +1,222 @@
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+from pathlib import Path
+import sys
+import unittest
+
+
+SCRIPTS = Path(__file__).resolve().parents[2] / "runtime" / "skill-optimizer" / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+from authoring.budget import (  # noqa: E402
+    AuthoringBudgetExceeded,
+    AuthoringBudgetLedger,
+    AuthoringBudgetLimits,
+    AuthoringCostKind,
+    validate_authoring_budget_record,
+)
+from core.canonical import digest_json  # noqa: E402
+
+
+def _resign(payload: dict) -> dict:
+    unsigned = {key: value for key, value in payload.items() if key != "content_digest"}
+    payload["content_digest"] = digest_json(unsigned)
+    return payload
+
+
+class AuthoringBudgetTests(unittest.TestCase):
+    def test_small_skill_does_not_spend_available_optional_budget(self) -> None:
+        ledger = AuthoringBudgetLedger(
+            AuthoringBudgetLimits(
+                research_calls=3,
+                source_checks=2,
+                independent_calls=2,
+                candidate_count=2,
+                revision_count=2,
+                allowed_modules=("counter-design", "source-research"),
+            )
+        )
+        self.assertEqual(0, ledger.used("research_call"))
+        self.assertEqual(0, ledger.used("source_check"))
+        ledger.record(AuthoringCostKind.CANDIDATE, reason="one minimal candidate")
+        ledger.stop("core_complete")
+        self.assertEqual(1, ledger.used("candidate"))
+        self.assertEqual(0, ledger.used("independent_call"))
+        self.assertEqual((), ledger.selected_modules)
+
+    def test_all_hard_counters_and_module_allowlist_are_enforced(self) -> None:
+        ledger = AuthoringBudgetLedger(
+            AuthoringBudgetLimits(
+                clarification_calls=1,
+                research_calls=1,
+                source_checks=1,
+                independent_calls=1,
+                candidate_count=1,
+                revision_count=1,
+                allowed_modules=("counter-design",),
+            )
+        )
+        ledger.record("clarification_call", reason="scope", source_ref="conversation:1")
+        ledger.record("research_call", reason="adjacent", source_ref="file:rules")
+        ledger.record("source_check", reason="verify", source_ref="file:rules")
+        ledger.record("independent_call", reason="counter", source_ref="trace:conflict")
+        ledger.select_module("counter-design", trigger_evidence=("trace:conflict",))
+        ledger.record("candidate", reason="draft")
+        ledger.record("revision", reason="one causal fix", source_ref="trace:failure")
+        with self.assertRaises(AuthoringBudgetExceeded):
+            ledger.record("revision", reason="second fix")
+        with self.assertRaises(ValueError):
+            ledger.select_module("unplanned", trigger_evidence=("trace:x",))
+        self.assertEqual(2, len(ledger.blocked_actions))
+
+    def test_blocked_actions_do_not_change_usage_and_are_replayed(self) -> None:
+        ledger = AuthoringBudgetLedger(AuthoringBudgetLimits(research_calls=1))
+        ledger.record("research_call", reason="first", source_ref="file:one")
+        with self.assertRaises(AuthoringBudgetExceeded) as caught:
+            ledger.record("research_call", reason="over", source_ref="file:two")
+        self.assertIsNotNone(caught.exception.blocked_action)
+        self.assertEqual(1, ledger.used("research_call"))
+        self.assertEqual(1, len(ledger.events))
+        self.assertEqual(1, len(ledger.blocked_actions))
+        rebuilt = validate_authoring_budget_record(ledger.to_dict())
+        self.assertEqual(ledger.canonical_digest, rebuilt.content_digest)
+
+    def test_stopped_ledger_records_blocked_attempt_without_new_action(self) -> None:
+        ledger = AuthoringBudgetLedger(AuthoringBudgetLimits(candidate_count=1))
+        ledger.stop("user_requested")
+        with self.assertRaises(AuthoringBudgetExceeded):
+            ledger.record("candidate", reason="late")
+        self.assertEqual(0, ledger.used("candidate"))
+        self.assertEqual(1, len(ledger.blocked_actions))
+
+    def test_selected_module_requires_trigger_evidence_and_is_frozen(self) -> None:
+        limits = AuthoringBudgetLimits(
+            allowed_modules=("counter-design", "source-research"),
+            max_selected_modules=1,
+        )
+        ledger = AuthoringBudgetLedger(limits)
+        with self.assertRaises(ValueError):
+            ledger.select_module("counter-design", trigger_evidence=())
+        with self.assertRaises(ValueError):
+            ledger.select_module("counter-design", trigger_evidence="trace:x")
+        with self.assertRaises(ValueError):
+            ledger.select_module("not-planned", trigger_evidence=("trace:x",))
+        ledger.select_module(
+            "counter-design",
+            trigger_evidence=("trace:ambiguous-design",),
+            source_refs=("trace:ambiguous-design",),
+        )
+        with self.assertRaises(AuthoringBudgetExceeded):
+            ledger.select_module("source-research", trigger_evidence=("trace:y",))
+        self.assertEqual(("counter-design", "source-research"), limits.allowed_modules)
+        # The immutable tuple cannot be expanded by mutating a caller list.
+        source = ["counter-design"]
+        frozen = AuthoringBudgetLimits(allowed_modules=source)
+        source.append("unplanned")
+        self.assertEqual(("counter-design",), frozen.allowed_modules)
+
+    def test_m_receipt_contains_no_quality_or_promotion_claim(self) -> None:
+        ledger = AuthoringBudgetLedger(AuthoringBudgetLimits(candidate_count=1))
+        ledger.record("candidate", reason="host wrote candidate")
+        ledger.stop("core_complete")
+        payload = ledger.to_dict()
+        serialized = json.dumps(payload, sort_keys=True)
+        for forbidden in (
+            "quality_pass",
+            "candidate_gain",
+            "promote",
+            "release",
+            "verified_claim",
+            "formal_outcome",
+        ):
+            self.assertNotIn(forbidden, serialized)
+
+    def test_tampered_usage_actions_digest_and_module_are_rejected(self) -> None:
+        ledger = AuthoringBudgetLedger(
+            AuthoringBudgetLimits(candidate_count=1, revision_count=1, allowed_modules=("m",))
+        )
+        ledger.record("candidate", reason="draft")
+        ledger.select_module("m", trigger_evidence=("trace:need",))
+        ledger.stop("done")
+        original = ledger.to_dict()
+
+        usage = deepcopy(original)
+        usage["usage"]["candidate"] = 0
+        with self.assertRaisesRegex(ValueError, "usage does not match|replayed actions"):
+            validate_authoring_budget_record(_resign(usage))
+
+        action = deepcopy(original)
+        action["actions"].append(
+            {
+                "sequence": 1,
+                "kind": "revision",
+                "amount": 2,
+                "reason": "forged",
+                "source_refs": [],
+                "trigger_evidence": [],
+            }
+        )
+        action["usage"]["revision"] = 2
+        with self.assertRaisesRegex(ValueError, "exceed|limit"):
+            validate_authoring_budget_record(_resign(action))
+
+        bad_digest = deepcopy(original)
+        bad_digest["content_digest"] = "sha256:" + "0" * 64
+        with self.assertRaisesRegex(ValueError, "digest mismatch"):
+            validate_authoring_budget_record(bad_digest)
+
+        bad_module = deepcopy(original)
+        bad_module["selected_modules"][0]["module_id"] = "unplanned"
+        with self.assertRaisesRegex(ValueError, "allowlist"):
+            validate_authoring_budget_record(_resign(bad_module))
+
+    def test_closed_record_rejects_claim_fields_even_when_resigned(self) -> None:
+        ledger = AuthoringBudgetLedger(AuthoringBudgetLimits(candidate_count=1))
+        ledger.stop("done")
+        forged = ledger.to_dict()
+        forged["candidate_gain"] = 1
+        with self.assertRaisesRegex(ValueError, "closed contract"):
+            validate_authoring_budget_record(_resign(forged))
+
+    def test_forged_blocked_action_must_replay_as_actually_blocked(self) -> None:
+        ledger = AuthoringBudgetLedger(
+            AuthoringBudgetLimits(allowed_modules=("counter-design",), max_selected_modules=1)
+        )
+        ledger.stop("done")
+        forged = ledger.to_dict()
+        forged["blocked_actions"] = [
+            {
+                "sequence": 0,
+                "kind": "module_selection",
+                "amount": 1,
+                "reason": "claimed block",
+                "limit": 1,
+                "used": 0,
+                "source_refs": [],
+                "trigger_evidence": ["trace:x"],
+                "module_id": "counter-design",
+            }
+        ]
+        with self.assertRaisesRegex(ValueError, "was allowed"):
+            validate_authoring_budget_record(_resign(forged))
+
+    def test_legacy_constructor_aliases_are_normalized(self) -> None:
+        limits = AuthoringBudgetLimits(
+            clarification_count=1,
+            research_count=2,
+            source_check_count=3,
+            counter_design_calls=1,
+            max_modules=1,
+            allowed_modules=("counter-design",),
+        )
+        self.assertEqual(1, limits.clarification_calls)
+        self.assertEqual(2, limits.research_calls)
+        self.assertEqual(3, limits.source_checks)
+        self.assertEqual(1, limits.independent_calls)
+        self.assertEqual(1, limits.max_selected_modules)
+
+
+if __name__ == "__main__":
+    unittest.main()

dev/optimizer-evals/test_authoring_provenance.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import Path
+import sys
+import unittest
+
+
+SCRIPTS = Path(__file__).resolve().parents[2] / "runtime" / "skill-optimizer" / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+from authoring.provenance import (  # noqa: E402
+    DevelopmentExample,
+    ProvenanceError,
+    ProvenanceLedger,
+    SourceKind,
+    SourceResolution,
+    validate_development_examples,
+    validate_no_holdout_overlap,
+)
+
+
+class AuthoringProvenanceTests(unittest.TestCase):
+    def test_only_four_source_kinds_are_accepted(self) -> None:
+        examples = validate_development_examples(
+            (
+                DevelopmentExample("e1", "observed", "conversation:12", "Real request"),
+                DevelopmentExample("e2", "user_confirmed", "user:confirmation-1", "Confirmed"),
+                DevelopmentExample("e3", "synthetic", "synthetic:boundary", "Generated"),
+                DevelopmentExample("e4", "assumed", "assumption:format", "Provisional"),
+            ),
+            source_resolver=lambda _kind, _ref: True,
+        )
+        self.assertEqual(
+            ["observed", "user_confirmed", "synthetic", "assumed"],
+            [item.source_kind.value for item in examples],
+        )
+        with self.assertRaises(ProvenanceError):
+            DevelopmentExample("bad", "inferred", "trace:x", "bad")
+
+    def test_trusted_kinds_require_resolver_and_prefixes_are_not_authority(self) -> None:
+        observed = DevelopmentExample("real", "observed", "synthetic:invented", "Observed")
+        with self.assertRaisesRegex(ProvenanceError, "resolver"):
+            validate_development_examples((observed,))
+        with self.assertRaisesRegex(ProvenanceError, "unresolved|unapproved"):
+            validate_development_examples((observed,), source_resolver=lambda _kind, _ref: False)
+        # An injected resolver, not a prefix heuristic, decides approval.
+        resolved = validate_development_examples(
+            (observed,),
+            source_resolver=lambda kind, ref: SourceResolution(kind, ref, True),
+        )
+        self.assertEqual(observed.source_ref, resolved[0].source_ref)
+
+    def test_resolver_mapping_must_be_closed_and_match_request(self) -> None:
+        example = DevelopmentExample("e", "user_confirmed", "user:event-1", "Confirmed")
+        malformed = {
+            "source_kind": "user_confirmed",
+            "source_ref": "user:event-1",
+            "approved": True,
+            "untrusted_note": "caller-authored",
+        }
+        with self.assertRaisesRegex(ProvenanceError, "closed contract"):
+            validate_development_examples((example,), source_resolver=lambda _k, _r: malformed)
+        mismatch = SourceResolution("observed", "conversation:other", True)
+        with self.assertRaisesRegex(ProvenanceError, "does not match"):
+            validate_development_examples((example,), source_resolver=lambda _k, _r: mismatch)
+
+    def test_internal_record_has_content_and_record_digests(self) -> None:
+        example = DevelopmentExample("e", "synthetic", "synthetic:one", "Generated case")
+        payload = example.to_dict()
+        self.assertIn("content_digest", payload)
+        self.assertIn("record_digest", payload)
+        projection = example.projection()
+        self.assertEqual({"source_kind", "source_ref"}, set(projection))
+        self.assertNotIn("summary", projection)
+        self.assertNotIn("content_digest", projection)
+        with self.assertRaisesRegex(ProvenanceError, "content_digest"):
+            DevelopmentExample(
+                "e",
+                "synthetic",
+                "synthetic:one",
+                "Generated case",
+                content_digest="sha256:" + "0" * 64,
+            )
+
+    def test_projection_is_strictly_two_fields_for_every_source_kind(self) -> None:
+        examples = tuple(
+            DevelopmentExample(str(index), kind, f"{kind}:ref", "summary")
+            for index, kind in enumerate(("observed", "user_confirmed", "synthetic", "assumed"))
+        )
+        projected = validate_development_examples(
+            examples,
+            source_resolver=lambda _kind, _ref: True,
+        )
+        for item in projected:
+            self.assertEqual({"source_kind", "source_ref"}, set(item.projection()))
+
+    def test_synthetic_and_assumed_never_become_holdout_or_representative(self) -> None:
+        synthetic = DevelopmentExample("s", "synthetic", "synthetic:one", "Generated")
+        assumed = DevelopmentExample("a", "assumed", "assumption:one", "Provisional")
+        self.assertFalse(synthetic.supports_holdout)
+        self.assertFalse(assumed.supports_representativeness)
+        with self.assertRaisesRegex(ProvenanceError, "overlaps a holdout"):
+            validate_no_holdout_overlap(
+                (synthetic,),
+                holdout_examples=(DevelopmentExample("h", "synthetic", "synthetic:one", "Holdout"),),
+            )
+        with self.assertRaisesRegex(ProvenanceError, "overlaps a holdout"):
+            validate_no_holdout_overlap((assumed,), holdout_refs=("assumption:one",))
+
+    def test_ledger_rejects_relabel_and_closes_projection_and_digests(self) -> None:
+        ledger = ProvenanceLedger(
+            source_resolver=lambda _kind, _ref: True,
+        )
+        ledger.add(DevelopmentExample("same", "synthetic", "synthetic:one", "Generated"))
+        with self.assertRaisesRegex(ProvenanceError, "relabelled"):
+            ledger.add(DevelopmentExample("same", "observed", "conversation:one", "Generated"))
+        payload = ledger.to_dict()
+        self.assertEqual(
+            [{"source_kind": "synthetic", "source_ref": "synthetic:one"}],
+            payload["handoff_projection"],
+        )
+        rebuilt = ProvenanceLedger.from_dict(payload)
+        self.assertEqual(ledger.record_digest, rebuilt.record_digest)
+        forged = deepcopy(payload)
+        forged["handoff_projection"][0]["summary"] = "claim"
+        with self.assertRaises(ProvenanceError):
+            ProvenanceLedger.from_dict(forged)
+
+    def test_mapping_inputs_and_holdout_overlap_are_checked(self) -> None:
+        raw = {
+            "example_id": "e",
+            "source_kind": "synthetic",
+            "source_ref": "synthetic:e",
+            "summary": "Generated",
+        }
+        normalized = validate_development_examples((raw,))
+        self.assertEqual("e", normalized[0].example_id)
+        with self.assertRaisesRegex(ProvenanceError, "overlaps a holdout"):
+            validate_development_examples((raw,), holdout_refs=("synthetic:e",))
+
+    def test_bool_resolver_is_only_an_injected_decision_not_a_source_field(self) -> None:
+        # A resolver may be a compatibility callable, but a provenance record
+        # cannot smuggle an approval boolean into its own mapping.
+        example = DevelopmentExample("e", "observed", "conversation:e", "Observed")
+        self.assertTrue(
+            validate_development_examples((example,), source_resolver=lambda _k, _r: True)
+        )
+        forged = example.to_dict()
+        forged["approved"] = True
+        with self.assertRaisesRegex(ProvenanceError, "fields do not match"):
+            DevelopmentExample.from_dict(forged)
+
+    def test_falsey_resolver_object_is_not_dropped_by_ledger(self) -> None:
+        class FalseyResolver:
+            def __bool__(self) -> bool:
+                return False
+
+            def __call__(self, kind, ref) -> bool:
+                return kind is SourceKind.OBSERVED and ref == "conversation:e"
+
+        ledger = ProvenanceLedger(source_resolver=FalseyResolver())
+        ledger.add(DevelopmentExample("e", "observed", "conversation:e", "Observed"))
+        self.assertEqual(1, len(ledger.examples))
+
+
+if __name__ == "__main__":
+    unittest.main()