feat: add Track C risk-aware delivery

dev/schema-tests/test_delivery_contracts.py

@@ -0,0 +1,380 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import replace
+import inspect
+import json
+from pathlib import Path
+import stat
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+
+WORKSPACE = Path(__file__).resolve().parents[2]
+SCRIPTS = WORKSPACE / "runtime" / "skill-optimizer" / "scripts"
+SCHEMAS = SCRIPTS / "schemas"
+sys.path.insert(0, str(SCRIPTS))
+
+from core.behavior_risk import (  # noqa: E402
+    BehaviorReportValidationError,
+    audit_behavior_risk,
+    report_from_dict,
+)
+from core.canonical import digest_json  # noqa: E402
+from core.delivery import (  # noqa: E402
+    DeliveryError,
+    DeliveryTarget,
+    build_behavior_and_delivery_summary,
+    evaluate_delivery_eligibility,
+)
+from core.workflow import (  # noqa: E402
+    AuthorizationGrant,
+    WorkflowActor,
+    WorkflowEvent,
+    WorkflowEventType,
+    WorkflowState,
+)
+import packaging.team_delivery as team_delivery  # noqa: E402
+from packaging.team_delivery import (  # noqa: E402
+    TeamDeliveryIntegrityError,
+    build_team_delivery,
+    prepare_team_delivery_action_target,
+    validate_team_delivery_manifest,
+)
+from validators.contracts import _validate_schema  # noqa: E402
+
+
+def _load_schema(filename: str) -> dict:
+    return json.loads((SCHEMAS / filename).read_text(encoding="utf-8"))
+
+
+def _schema_errors(filename: str, document: dict) -> list[str]:
+    schema = _load_schema(filename)
+    return _validate_schema(document, schema, schema)
+
+
+def _write_safe_skill(root: Path) -> Path:
+    candidate = root / "candidate"
+    candidate.mkdir()
+    (candidate / "SKILL.md").write_text(
+        "---\n"
+        "name: quiet-summary\n"
+        "description: Summarize supplied prose without side effects.\n"
+        "---\n"
+        "# Quiet summary\n\n"
+        "Return a concise summary of the supplied prose.\n",
+        encoding="utf-8",
+    )
+    return candidate
+
+
+def _authorization_event(
+    *,
+    workflow_id: str,
+    scope_digest: str,
+    risk_digest: str,
+    target_digest: str,
+    kind: str,
+    action: str,
+) -> WorkflowEvent:
+    grant = AuthorizationGrant(
+        authorization_id=f"{workflow_id}-authorization",
+        kind=kind,
+        status="granted",
+        actions=(action, "quality_unverified"),
+        scope_digest=scope_digest,
+        risk_digest=risk_digest,
+        target_digest=target_digest,
+        expires_at="2099-01-01T00:00:00Z",
+    )
+    return WorkflowEvent.create(
+        sequence=0,
+        workflow_id=workflow_id,
+        event_type=WorkflowEventType.AUTHORIZATION,
+        from_state=WorkflowState.INTAKE,
+        to_state=WorkflowState.INTAKE,
+        actor=WorkflowActor.USER,
+        payload={"grant": grant.to_dict()},
+        created_at="2026-07-26T00:00:00+00:00",
+        previous_event_digest=None,
+    )
+
+
+class BehaviorSchemaRuntimeParityTests(unittest.TestCase):
+    def test_behavior_positive_fixture_passes_schema_and_runtime(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            candidate = _write_safe_skill(Path(temporary))
+            report = audit_behavior_risk(candidate).to_dict()
+            self.assertEqual([], _schema_errors("behavior-risk-report.schema.json", report))
+            self.assertEqual(report, report_from_dict(report).to_dict())
+
+    def test_behavior_closed_negative_fixture_hits_both_surfaces(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            candidate = _write_safe_skill(Path(temporary))
+            report = audit_behavior_risk(candidate).to_dict()
+            report["runtime_safe"] = True
+            self.assertTrue(_schema_errors("behavior-risk-report.schema.json", report))
+            with self.assertRaises(BehaviorReportValidationError):
+                report_from_dict(report)
+
+    def test_behavior_semantic_digest_and_rck_are_rebuilt_at_runtime(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            candidate = _write_safe_skill(Path(temporary))
+            report = audit_behavior_risk(candidate).to_dict()
+            tampered = deepcopy(report)
+            tampered["mandatory_controls"] = ["caller-selected-control"]
+            unsigned = {key: value for key, value in tampered.items() if key != "content_digest"}
+            tampered["content_digest"] = digest_json(unsigned)
+            # Shape validation is intentionally not treated as semantic proof.
+            self.assertEqual(
+                [], _schema_errors("behavior-risk-report.schema.json", tampered)
+            )
+            with self.assertRaisesRegex(
+                BehaviorReportValidationError, "mandatory_controls"
+            ):
+                report_from_dict(tampered)
+
+
+class DeliveryTrustBoundaryTests(unittest.TestCase):
+    def _eligible_personal(self, root: Path, *, quality_summary=None):
+        candidate = _write_safe_skill(root)
+        personal_root = root / ".codex" / "skills"
+        personal_root.mkdir(parents=True)
+        target = personal_root / "quiet-summary"
+        report = audit_behavior_risk(candidate)
+        scope_digest = digest_json({"scope": "personal-test"})
+        eligibility = evaluate_delivery_eligibility(
+            candidate,
+            delivery_target=DeliveryTarget.PERSONAL_INSTALL,
+            target=target,
+            approved_root=personal_root,
+            scope_digest=scope_digest,
+            workflow_id="delivery-personal",
+            host="codex",
+            quality_summary=quality_summary,
+        )
+        return eligibility, report
+
+    def test_generic_evaluator_blocks_without_trusted_workflow_source(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            eligibility, report = self._eligible_personal(Path(temporary))
+            self.assertFalse(eligibility.eligible)
+            self.assertTrue(eligibility.quality_unverified)
+            self.assertEqual(report.minimum_risk, eligibility.minimum_risk)
+            self.assertEqual(report.mandatory_controls, eligibility.mandatory_controls)
+            self.assertEqual((), eligibility.quality_projection.verified_claims)
+            self.assertIn(
+                "trusted-workflow-event-source-adapter",
+                eligibility.integration_requests,
+            )
+            self.assertNotIn(
+                "workflow_events",
+                inspect.signature(evaluate_delivery_eligibility).parameters,
+            )
+
+    def test_caller_verified_and_eligible_fields_never_upgrade_claims(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            eligibility, _ = self._eligible_personal(
+                Path(temporary),
+                quality_summary={
+                    "verified_claims": ["formal-adoption"],
+                    "unverified_claims": [],
+                    "blocked_claims": [],
+                    "removed_claims": [],
+                    "eligible": True,
+                },
+            )
+            self.assertFalse(eligibility.eligible)
+            self.assertEqual((), eligibility.quality_projection.verified_claims)
+            self.assertIn(
+                "formal-adoption", eligibility.quality_projection.blocked_claims
+            )
+            self.assertIn(
+                "reject-caller-authored-verified-claims",
+                eligibility.integration_requests,
+            )
+
+    def test_required_claim_and_automatic_routing_fail_closed(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            candidate = _write_safe_skill(root)
+            output = root / "approved-team-output"
+            output.mkdir()
+            report = audit_behavior_risk(candidate)
+            scope_digest = digest_json({"scope": "team-test"})
+            eligibility = evaluate_delivery_eligibility(
+                candidate,
+                delivery_target=DeliveryTarget.TEAM_PACKAGE,
+                target=output,
+                approved_root=output,
+                scope_digest=scope_digest,
+                workflow_id="delivery-team",
+                host="codex",
+                required_claims=("formal-adoption",),
+            )
+            self.assertFalse(eligibility.eligible)
+            self.assertIn(
+                "gate1-quality-raw-graph-adapter",
+                eligibility.integration_requests,
+            )
+            self.assertIn(
+                "automatic-routing-scope-adapter",
+                eligibility.integration_requests,
+            )
+
+    def test_forged_typed_eligibility_and_receipt_projection_are_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            eligibility, report = self._eligible_personal(Path(temporary))
+            with self.assertRaisesRegex(ValueError, "eligible=true"):
+                replace(eligibility, eligible=True, reasons=())
+            with self.assertRaisesRegex(DeliveryError, "receipt projection"):
+                build_behavior_and_delivery_summary(
+                    report,
+                    eligibility,
+                    delivery_receipt={"status": "installed", "forged": True},
+                )
+
+    def test_delivery_rejects_candidate_root_symlink(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            candidate = _write_safe_skill(root)
+            link = root / "candidate-link"
+            link.symlink_to(candidate, target_is_directory=True)
+            personal_root = root / ".codex" / "skills"
+            personal_root.mkdir(parents=True)
+            with self.assertRaisesRegex(DeliveryError, "symbolic link"):
+                evaluate_delivery_eligibility(
+                    link,
+                    delivery_target=DeliveryTarget.PERSONAL_INSTALL,
+                    target=personal_root / "quiet-summary",
+                    approved_root=personal_root,
+                    scope_digest=digest_json({"scope": "symlink"}),
+                )
+
+
+class TeamManifestSchemaRuntimeParityTests(unittest.TestCase):
+    def _build_manifest(self, root: Path):
+        candidate = _write_safe_skill(root)
+        output = root / "approved-team-output"
+        output.mkdir()
+        scope_digest = digest_json({"workflow_scope": "explicit-team"})
+        target = prepare_team_delivery_action_target(
+            candidate,
+            output,
+            host="codex",
+            scope_digest=scope_digest,
+        )
+        event = _authorization_event(
+            workflow_id="schema-team-delivery",
+            scope_digest=scope_digest,
+            risk_digest=target.risk_commitment_digest,
+            target_digest=target.content_digest,
+            kind="external_write",
+            action="team_delivery",
+        )
+        workflow_root = root / "trusted-codex-workflows"
+        workflow_root.mkdir(mode=0o700)
+        workflow_root.chmod(0o700)
+        workflow_path = workflow_root / team_delivery._workflow_event_filename(
+            event.workflow_id
+        )
+        workflow_path.write_text(
+            json.dumps(event.to_dict(), ensure_ascii=True, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        workflow_path.chmod(0o600)
+        with mock.patch.object(
+            team_delivery, "_codex_workflow_root", return_value=workflow_root
+        ):
+            manifest = build_team_delivery(
+                candidate,
+                output,
+                host="codex",
+                workflow_id=event.workflow_id,
+                scope_digest=scope_digest,
+                risk_digest=target.risk_commitment_digest,
+                expected_candidate_digest=target.candidate_digest,
+            )
+        return manifest, output
+
+    def _validate_manifest(self, manifest):
+        output_root = (
+            Path(manifest.output_root)
+            if hasattr(manifest, "output_root")
+            else Path(manifest["output_root"])
+        )
+        workflow_root = output_root.parent / "trusted-codex-workflows"
+        with mock.patch.object(
+            team_delivery, "_codex_workflow_root", return_value=workflow_root
+        ):
+            return validate_team_delivery_manifest(manifest)
+
+    def test_team_manifest_positive_fixture_passes_schema_and_current_bytes(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            manifest, _output = self._build_manifest(Path(temporary))
+            document = manifest.to_dict()
+            self.assertEqual(
+                [], _schema_errors("team-delivery-manifest.schema.json", document)
+            )
+            self.assertEqual(
+                manifest.content_digest,
+                self._validate_manifest(document).content_digest,
+            )
+
+    def test_team_manifest_negative_fixture_hits_schema_and_runtime(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            manifest, _output = self._build_manifest(Path(temporary))
+            forged = deepcopy(manifest.to_dict())
+            forged["compatibility"]["status"] = "verified"
+            forged["content_digest"] = digest_json(
+                {key: value for key, value in forged.items() if key != "content_digest"}
+            )
+            self.assertTrue(
+                _schema_errors("team-delivery-manifest.schema.json", forged)
+            )
+            with self.assertRaises(ValueError):
+                self._validate_manifest(forged)
+
+    def test_team_manifest_runtime_rechecks_package_bytes(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            manifest, output = self._build_manifest(Path(temporary))
+            skill = output / "package" / "SKILL.md"
+            skill.chmod(stat.S_IMODE(skill.stat().st_mode) | stat.S_IWUSR)
+            skill.write_text("tampered\n", encoding="utf-8")
+            with self.assertRaises(TeamDeliveryIntegrityError):
+                self._validate_manifest(manifest.to_dict())
+
+    def test_current_byte_risk_cannot_be_lowered_by_delivery_target(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            candidate = root / "candidate"
+            candidate.mkdir()
+            (candidate / "SKILL.md").write_text(
+                "---\nname: remote-writer\ndescription: Publish a report.\n---\n"
+                "# Remote writer\n\nPublish and upload the report through a network API.\n",
+                encoding="utf-8",
+            )
+            personal_root = root / ".codex" / "skills"
+            personal_root.mkdir(parents=True)
+            target = personal_root / "remote-writer"
+            report = audit_behavior_risk(candidate)
+            eligibility = evaluate_delivery_eligibility(
+                candidate,
+                delivery_target=DeliveryTarget.PERSONAL_INSTALL,
+                target=target,
+                approved_root=personal_root,
+                scope_digest=digest_json({"scope": "risk-test"}),
+            )
+            self.assertGreaterEqual(report.minimum_risk.severity, 2)
+            self.assertEqual(report.minimum_risk, eligibility.minimum_risk)
+            self.assertEqual(report.mandatory_controls, eligibility.mandatory_controls)
+            self.assertEqual(
+                report.mandatory_capabilities, eligibility.mandatory_capabilities
+            )
+            self.assertFalse(eligibility.eligible)
+
+
+if __name__ == "__main__":
+    unittest.main()