diff --git a/.github/workflows/health-69-consumer-sync-shadow-evidence.yml b/.github/workflows/health-69-consumer-sync-shadow-evidence.yml index dc49c86c4..c3c7abdfc 100644 --- a/.github/workflows/health-69-consumer-sync-shadow-evidence.yml +++ b/.github/workflows/health-69-consumer-sync-shadow-evidence.yml @@ -8,6 +8,10 @@ on: - '.github/sync-manifest.yml' - 'scripts/sync_manifest_compiler.py' - 'scripts/build_consumer_sync_shadow_handoff.py' + - 'scripts/completion_evidence_adapter.py' + - 'scripts/orchestrator_runtime/**' + - 'config/orchestrator_runtime/**' + - 'tests/orchestrator_runtime/**' - 'templates/consumer-repo/**' schedule: - cron: '17 8 * * 1' @@ -18,6 +22,10 @@ on: - '.github/sync-manifest.yml' - 'scripts/sync_manifest_compiler.py' - 'scripts/build_consumer_sync_shadow_handoff.py' + - 'scripts/completion_evidence_adapter.py' + - 'scripts/orchestrator_runtime/**' + - 'config/orchestrator_runtime/**' + - 'tests/orchestrator_runtime/**' - 'templates/consumer-repo/**' permissions: @@ -62,6 +70,13 @@ jobs: --run-ref "$SHADOW_RUN_REF" \ --output consumer-sync-shadow-evidence/handoff.json + - name: Validate and record typed completion evidence + run: | + python -m scripts.orchestrator_runtime.completion_event_adapter \ + --handoff consumer-sync-shadow-evidence/handoff.json \ + --output consumer-sync-shadow-evidence/completion-evidence.json \ + --state-dir consumer-sync-shadow-evidence + - name: Upload shadow evidence uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: @@ -78,6 +93,9 @@ jobs: handoff = json.loads( Path('consumer-sync-shadow-evidence/handoff.json').read_text() ) + completion_evidence = json.loads( + Path('consumer-sync-shadow-evidence/completion-evidence.json').read_text() + ) print('## Consumer sync shadow evidence') print(f"- Plan: {handoff['plan_id']}") print(f"- Copy entries: {handoff['entry_count']}") @@ -85,4 +103,5 @@ jobs: print('- Supervision: shadow') print('- Write authority: false') print('- Promotion allowed: false') + print(f"- Completion evidence: {completion_evidence['status']}") PY diff --git a/config/orchestrator_runtime/capabilities.json b/config/orchestrator_runtime/capabilities.json new file mode 100644 index 000000000..336032d39 --- /dev/null +++ b/config/orchestrator_runtime/capabilities.json @@ -0,0 +1,6 @@ +{ + "capability:reference-sync-hygiene-test-gate": { + "counterexamples": [], + "lifecycle": "shadow" + } +} diff --git a/docs/ci/WORKFLOWS.md b/docs/ci/WORKFLOWS.md index a6ef08af3..3ae064568 100644 --- a/docs/ci/WORKFLOWS.md +++ b/docs/ci/WORKFLOWS.md @@ -191,7 +191,7 @@ Scheduled health jobs keep the automation ecosystem aligned: * [`health-53-scorecard.yml`](../../.github/workflows/health-53-scorecard.yml) runs the OpenSSF Scorecard supply-chain posture scan (report-only/advisory) and uploads findings to code scanning (push to main, `branch_protection_rule`, weekly schedule, manual dispatch). The weekly repo-review coordinator also consumes the published Scorecard API results as a human-gated supplemental candidate source (`scorecard-scan.json`). * [`health-67-integration-sync-check.yml`](../../.github/workflows/health-67-integration-sync-check.yml) validates that Workflows-Integration-Tests repo stays in sync with templates (push, `repository_dispatch`, daily schedule). * [`health-68-consumer-sync-drift.yml`](../../.github/workflows/health-68-consumer-sync-drift.yml) detects drift in registered consumer repos (template/manifest changes, daily schedule, manual dispatch). -* [`health-69-consumer-sync-shadow-evidence.yml`](../../.github/workflows/health-69-consumer-sync-shadow-evidence.yml) publishes the typed plan and a read-only, non-promoting handoff artifact for the existing local Orchestrator shadow capability (weekly schedule, main changes, manual dispatch). +* [`health-69-consumer-sync-shadow-evidence.yml`](../../.github/workflows/health-69-consumer-sync-shadow-evidence.yml) publishes the typed plan, a read-only non-promoting handoff artifact, and a typed `completion-evidence.json` record produced by the versioned `scripts/orchestrator_runtime/` rail (weekly schedule, main changes, manual dispatch). * [`health-70-validate-sync-manifest.yml`](../../.github/workflows/health-70-validate-sync-manifest.yml) validates that sync-manifest.yml is complete - ensures all sync-able files are declared (PR, push). * [`health-71-sync-health-check.yml`](../../.github/workflows/health-71-sync-health-check.yml) monitors sync workflow health daily - creates issues if all recent runs failed or sync is stale (daily schedule, manual dispatch). * [`health-72-template-sync.yml`](../../.github/workflows/health-72-template-sync.yml) validates that manifest-declared exact template-sync files are in sync with their consumer template copies (PR, push on exact-sync source/template changes). diff --git a/docs/ci/WORKFLOW_SYSTEM.md b/docs/ci/WORKFLOW_SYSTEM.md index 7a51674b3..afc613b77 100644 --- a/docs/ci/WORKFLOW_SYSTEM.md +++ b/docs/ci/WORKFLOW_SYSTEM.md @@ -727,7 +727,7 @@ Keep this table handy when you are triaging automation: it confirms which workfl | **Health 53 Scorecard** (`health-53-scorecard.yml`, maintenance bucket) | `push` (main), `branch_protection_rule`, `schedule` (weekly), `workflow_dispatch` | Run the OpenSSF Scorecard supply-chain posture scan (report-only/advisory) and upload findings to code scanning. Published API results are consumed by the weekly repo-review Scorecard scan (`scorecard-scan.json`) as a human-gated candidate source. | ⚪ Automatic/scheduled | [Scorecard runs](https://github.com/stranske/Workflows/actions/workflows/health-53-scorecard.yml) | | **Health 67 Integration Sync Check** (`health-67-integration-sync-check.yml`, maintenance bucket) | `push` (templates), `repository_dispatch`, `schedule` (daily) | Validate that Workflows-Integration-Tests repo stays in sync with templates. Creates issues when drift detected. | ⚪ Automatic/scheduled | [Integration sync runs](https://github.com/stranske/Workflows/actions/workflows/health-67-integration-sync-check.yml) | | **Health 68 Consumer Sync Drift Check** (`health-68-consumer-sync-drift.yml`, maintenance bucket) | `push` (templates + manifest + scripts), `schedule` (daily), `workflow_dispatch` | Detect template/manifest drift across registered consumer repos. Creates or updates issues when drift is detected. | ⚪ Automatic/scheduled | [Consumer drift runs](https://github.com/stranske/Workflows/actions/workflows/health-68-consumer-sync-drift.yml) | -| **Health 69 Consumer Sync Shadow Evidence** (`health-69-consumer-sync-shadow-evidence.yml`, maintenance bucket) | `push` (manifest, compiler, handoff producer, or consumer templates), `schedule` (weekly), `workflow_dispatch` | Compile the typed sync plan and publish a bounded read-only Orchestrator handoff artifact. The workflow has no consumer write authority and cannot promote the capability. | ⚪ Automatic/scheduled/manual | [Consumer sync shadow evidence runs](https://github.com/stranske/Workflows/actions/workflows/health-69-consumer-sync-shadow-evidence.yml) | +| **Health 69 Consumer Sync Shadow Evidence** (`health-69-consumer-sync-shadow-evidence.yml`, maintenance bucket) | `push` (manifest, compiler, handoff producer, orchestrator runtime rail, or consumer templates), `schedule` (weekly), `workflow_dispatch` | Compile the typed sync plan, publish a bounded read-only Orchestrator handoff artifact, and record typed completion evidence via `scripts/orchestrator_runtime/completion_event_adapter.py`. The uploaded bundle includes `handoff.json`, `completion-evidence.json`, `capabilities-state.json`, and `evidence-ledger.json`. `completion-evidence.json` uses schema `workflows.runner-completion-evidence/v1`; `status` is `accepted`, `duplicate`, or `rejected`, with `diagnostic_code` explaining rejections. The workflow has no consumer write authority and cannot promote the capability. | ⚪ Automatic/scheduled/manual | [Consumer sync shadow evidence runs](https://github.com/stranske/Workflows/actions/workflows/health-69-consumer-sync-shadow-evidence.yml) | | **Health 70 Validate Sync Manifest** (`health-70-validate-sync-manifest.yml`, maintenance bucket) | `pull_request`, `push` | Validate that sync-manifest.yml includes all sync-able files. Fails PRs that add workflows/prompts/scripts without updating manifest. | ⚪ Required on PRs | [Manifest validation runs](https://github.com/stranske/Workflows/actions/workflows/health-70-validate-sync-manifest.yml) | | **Health 71 Sync Health Check** (`health-71-sync-health-check.yml`, maintenance bucket) | `schedule` (daily), `workflow_dispatch` | Monitor sync workflow health and create issues when all recent runs failed or sync is stale. | ⚪ Scheduled/manual | [Sync health check runs](https://github.com/stranske/Workflows/actions/workflows/health-71-sync-health-check.yml) | | **Health 72 Template Sync** (`health-72-template-sync.yml`, maintenance bucket) | `pull_request`, `push` (exact-sync sources/templates) | Validate that manifest-declared exact template-sync files match their `templates/consumer-repo/` copies. | ⚪ Required on PRs | [Template sync validation runs](https://github.com/stranske/Workflows/actions/workflows/health-72-template-sync.yml) | diff --git a/docs/ops/CONSUMER_REPO_MAINTENANCE.md b/docs/ops/CONSUMER_REPO_MAINTENANCE.md index 525d410c7..812daf879 100644 --- a/docs/ops/CONSUMER_REPO_MAINTENANCE.md +++ b/docs/ops/CONSUMER_REPO_MAINTENANCE.md @@ -266,6 +266,14 @@ handoff is explicitly `shadow`, `write_authority=false`, and and promotion blockers remain owned by Orchestrator's `consumer_sync_shadow.py` dashboard. +The same workflow also records typed completion evidence through +`scripts/orchestrator_runtime/completion_event_adapter.py`. The uploaded bundle +includes `completion-evidence.json` (`workflows.runner-completion-evidence/v1`) +plus mutable runtime state files `capabilities-state.json` and +`evidence-ledger.json`. Accepted evidence attaches only to capabilities present +in `config/orchestrator_runtime/capabilities.json`; duplicate replays return +`status=duplicate` without mutating ledger or capability state. + #### Canary-Gated Fan-out Maint 68 separates a sync plan into `preview`, `canary`, and `promote` phases. diff --git a/renovate-presets/consumer-managed-paths.json b/renovate-presets/consumer-managed-paths.json index 5f0e448d1..1a2f92a40 100644 --- a/renovate-presets/consumer-managed-paths.json +++ b/renovate-presets/consumer-managed-paths.json @@ -3,7 +3,7 @@ "description": "GENERATED by scripts/generate_consumer_renovate_ownership.py -- do not edit by hand. Disables Renovate dependency extraction for the paths that maint-68-sync-consumer-repos.yml overwrites from .github/sync-manifest.yml, and only in the consumer repos where that overwrite actually applies. Without this boundary a consumer's Renovate opens PRs against centrally-copied files (Inv-Man-Intake#838, Manager-Database#1347) that the next sync silently reverts. Renovate stays enabled for create-only/skipped paths the consumer owns, and for every canonical source file in stranske/Workflows, which is the sync source rather than a consumer. Regenerate with `python scripts/generate_consumer_renovate_ownership.py`; `--check` fails on drift and runs in scripts/dev_check.sh.", "packageRules": [ { - "description": "Maint 68 overwrites these 210 manifest-managed paths in every registered consumer; Renovate edits there are reverted on the next sync.", + "description": "Maint 68 overwrites these 211 manifest-managed paths in every registered consumer; Renovate edits there are reverted on the next sync.", "matchRepositories": [ "stranske/Collab-Admin", "stranske/Counter_Risk", @@ -82,6 +82,7 @@ ".github/scripts/issue_format.py", ".github/scripts/issue_pr_locator.js", ".github/scripts/issue_scope_parser.js", + ".github/scripts/keepalive_challenge_due.js", ".github/scripts/keepalive_contract.js", ".github/scripts/keepalive_gate.js", ".github/scripts/keepalive_guard_utils.js", diff --git a/scripts/completion_evidence_adapter.py b/scripts/completion_evidence_adapter.py new file mode 100644 index 000000000..4e5031372 --- /dev/null +++ b/scripts/completion_evidence_adapter.py @@ -0,0 +1,20 @@ +"""Backward-compatible re-exports for the versioned orchestrator runtime rail.""" + +from __future__ import annotations + +from scripts.orchestrator_runtime.capability_lifecycle import ingest_completion_evidence +from scripts.orchestrator_runtime.completion_event_adapter import main +from scripts.orchestrator_runtime.evidence_schema import SCHEMA +from scripts.orchestrator_runtime.runner_effect_bridge import completion_payload_from_shadow_handoff + +CompletionEvidenceError = __import__( + "scripts.orchestrator_runtime.evidence_schema", fromlist=["CompletionEvidenceError"] +).CompletionEvidenceError + +__all__ = [ + "CompletionEvidenceError", + "SCHEMA", + "completion_payload_from_shadow_handoff", + "ingest_completion_evidence", + "main", +] diff --git a/scripts/orchestrator_runtime/__init__.py b/scripts/orchestrator_runtime/__init__.py new file mode 100644 index 000000000..d96c3bfcf --- /dev/null +++ b/scripts/orchestrator_runtime/__init__.py @@ -0,0 +1,11 @@ +"""Versioned orchestrator runtime rail for completion-event evidence handling.""" + +from scripts.orchestrator_runtime.capability_lifecycle import ingest_completion_evidence +from scripts.orchestrator_runtime.completion_event_adapter import main as completion_event_main +from scripts.orchestrator_runtime.runner_effect_bridge import completion_payload_from_shadow_handoff + +__all__ = [ + "completion_event_main", + "completion_payload_from_shadow_handoff", + "ingest_completion_evidence", +] diff --git a/scripts/orchestrator_runtime/capabilities.py b/scripts/orchestrator_runtime/capabilities.py new file mode 100644 index 000000000..c85722aa8 --- /dev/null +++ b/scripts/orchestrator_runtime/capabilities.py @@ -0,0 +1,53 @@ +"""Authoritative capability registry for orchestrator evidence ingestion.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path +from typing import Any + + +class CapabilityRegistry: + """Load and persist the authoritative candidate/shadow capability registry.""" + + def __init__(self, capabilities: dict[str, dict[str, Any]]) -> None: + self._capabilities = deepcopy(capabilities) + + @classmethod + def load(cls, path: Path) -> CapabilityRegistry: + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError("capability registry must be an object") + capabilities: dict[str, dict[str, Any]] = {} + for capability_id, record in raw.items(): + if not isinstance(capability_id, str) or not isinstance(record, dict): + raise ValueError("capability registry entries must be capability-id objects") + lifecycle = record.get("lifecycle") + if lifecycle not in {"candidate", "shadow"}: + raise ValueError(f"unsupported lifecycle for {capability_id}") + counterexamples = record.get("counterexamples", []) + if not isinstance(counterexamples, list): + raise ValueError(f"counterexamples for {capability_id} must be a list") + capabilities[capability_id] = { + "lifecycle": lifecycle, + "counterexamples": list(counterexamples), + } + return cls(capabilities) + + def snapshot(self) -> dict[str, dict[str, Any]]: + return deepcopy(self._capabilities) + + def get(self, capability_id: str) -> dict[str, Any] | None: + record = self._capabilities.get(capability_id) + return deepcopy(record) if isinstance(record, dict) else None + + def save(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(self._capabilities, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + def apply_mutation(self, capabilities: dict[str, dict[str, Any]]) -> None: + self._capabilities = deepcopy(capabilities) diff --git a/scripts/orchestrator_runtime/capability_lifecycle.py b/scripts/orchestrator_runtime/capability_lifecycle.py new file mode 100644 index 000000000..bdb5cf8b6 --- /dev/null +++ b/scripts/orchestrator_runtime/capability_lifecycle.py @@ -0,0 +1,108 @@ +"""Evidence ledger persistence and idempotent capability association.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path +from typing import Any + +from scripts.orchestrator_runtime.evidence_schema import ( + CompletionEvidenceError, + stable_evidence_id, + validated_payload, +) + + +class EvidenceLedger: + """Durable evidence ledger persisted across adapter invocations.""" + + def __init__(self, records: list[dict[str, Any]] | None = None) -> None: + self._records = deepcopy(records or []) + + @classmethod + def load(cls, path: Path) -> EvidenceLedger: + if not path.is_file(): + return cls([]) + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, list): + raise ValueError("evidence ledger must be a list") + return cls(raw) + + def snapshot(self) -> list[dict[str, Any]]: + return deepcopy(self._records) + + def save(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(self._records, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + + def apply_mutation(self, records: list[dict[str, Any]]) -> None: + self._records = deepcopy(records) + + +def ingest_completion_evidence( + payload: Any, + *, + capabilities: dict[str, dict[str, Any]], + ledger: list[dict[str, Any]], +) -> dict[str, Any]: + """Return an idempotent ledger association without unsafe state transitions.""" + original_capabilities = deepcopy(capabilities) + original_ledger = deepcopy(ledger) + try: + evidence = validated_payload(payload) + capability = capabilities.get(evidence["capability_id"]) + if not isinstance(capability, dict) or capability.get("lifecycle") not in { + "candidate", + "shadow", + }: + raise CompletionEvidenceError( + "spoofed_capability_id: target must be an existing candidate or shadow" + ) + evidence_id = stable_evidence_id(evidence) + if any(record.get("evidence_id") == evidence_id for record in ledger): + return { + "status": "duplicate", + "diagnostic_code": "duplicate_evidence", + "evidence_id": evidence_id, + "capabilities": original_capabilities, + "ledger": original_ledger, + } + updated_capabilities = deepcopy(capabilities) + updated_ledger = deepcopy(ledger) + target = updated_capabilities[evidence["capability_id"]] + existing_counterexamples = list(target.get("counterexamples") or []) + for counterexample in evidence["counterexamples"]: + if counterexample not in existing_counterexamples: + existing_counterexamples.append(counterexample) + target["counterexamples"] = existing_counterexamples + record = { + "evidence_id": evidence_id, + "capability_id": evidence["capability_id"], + "lifecycle": target["lifecycle"], + "effect_fingerprint": evidence["effect_fingerprint"], + "evidence_artifact_ref": evidence["evidence_artifact_ref"], + "supervision_mode": evidence["supervision_mode"], + "terminal_disposition": evidence["terminal_disposition"], + "provenance": evidence["provenance"], + "counterexamples": evidence["counterexamples"], + } + updated_ledger.append(record) + return { + "status": "accepted", + "diagnostic_code": "accepted_evidence", + "evidence_id": evidence_id, + "capabilities": updated_capabilities, + "ledger": updated_ledger, + } + except CompletionEvidenceError as exc: + code, _, message = str(exc).partition(": ") + return { + "status": "rejected", + "diagnostic_code": code, + "message": message, + "capabilities": original_capabilities, + "ledger": original_ledger, + } diff --git a/scripts/orchestrator_runtime/completion_event_adapter.py b/scripts/orchestrator_runtime/completion_event_adapter.py new file mode 100644 index 000000000..5cf711979 --- /dev/null +++ b/scripts/orchestrator_runtime/completion_event_adapter.py @@ -0,0 +1,88 @@ +"""CLI entry point for Health 69 and orchestrator completion-event ingestion.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from scripts.orchestrator_runtime.capabilities import CapabilityRegistry +from scripts.orchestrator_runtime.capability_lifecycle import ( + EvidenceLedger, + ingest_completion_evidence, +) +from scripts.orchestrator_runtime.evidence_schema import CompletionEvidenceError, reject +from scripts.orchestrator_runtime.runner_effect_bridge import completion_payload_from_shadow_handoff + +DEFAULT_REGISTRY = Path("config/orchestrator_runtime/capabilities.json") + + +def _load_runtime_capabilities(registry_path: Path, state_path: Path) -> CapabilityRegistry: + if state_path.is_file(): + return CapabilityRegistry.load(state_path) + registry = CapabilityRegistry.load(registry_path) + registry.save(state_path) + return registry + + +def process_shadow_handoff( + handoff: dict[str, object], + *, + registry: CapabilityRegistry, + ledger: EvidenceLedger, +) -> dict[str, object]: + payload = completion_payload_from_shadow_handoff(handoff) + result = ingest_completion_evidence( + payload, + capabilities=registry.snapshot(), + ledger=ledger.snapshot(), + ) + if result["status"] == "accepted": + registry.apply_mutation(result["capabilities"]) + ledger.apply_mutation(result["ledger"]) + return result + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--handoff", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument( + "--registry", + type=Path, + default=DEFAULT_REGISTRY, + help="Authoritative capability registry JSON path", + ) + parser.add_argument( + "--state-dir", + type=Path, + help="Mutable runtime state directory for capabilities and ledger persistence", + ) + parser.add_argument( + "--ledger", + type=Path, + help="Persistent evidence ledger JSON path (defaults to /evidence-ledger.json)", + ) + args = parser.parse_args(argv) + state_dir = args.state_dir or args.output.parent + state_dir.mkdir(parents=True, exist_ok=True) + capabilities_state_path = state_dir / "capabilities-state.json" + ledger_path = args.ledger or (state_dir / "evidence-ledger.json") + try: + handoff = json.loads(args.handoff.read_text(encoding="utf-8")) + registry = _load_runtime_capabilities(args.registry, capabilities_state_path) + ledger = EvidenceLedger.load(ledger_path) + result = process_shadow_handoff(handoff, registry=registry, ledger=ledger) + if result["status"] != "accepted": + raise reject(result["diagnostic_code"], result.get("message", "handoff rejected")) + registry.save(capabilities_state_path) + ledger.save(ledger_path) + except (OSError, json.JSONDecodeError, CompletionEvidenceError, ValueError) as exc: + parser.error(str(exc)) + args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/orchestrator_runtime/evidence_schema.py b/scripts/orchestrator_runtime/evidence_schema.py new file mode 100644 index 000000000..845333916 --- /dev/null +++ b/scripts/orchestrator_runtime/evidence_schema.py @@ -0,0 +1,117 @@ +"""Schema and validation for runner completion evidence payloads.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from scripts.runner_lib import normalize_capability_effect_evidence + +SCHEMA = "workflows.runner-completion-evidence/v1" +ALLOWED_KEYS = frozenset( + { + "schema", + "capability_id", + "effect_fingerprint", + "evidence_artifact_ref", + "supervision_mode", + "capability_evidence_status", + "terminal_disposition", + "provenance", + "counterexamples", + } +) +MAX_COUNTEREXAMPLES = 32 +MAX_COUNTEREXAMPLE_LENGTH = 512 + + +class CompletionEvidenceError(ValueError): + """A rejection that must not mutate capability state or the ledger.""" + + +def reject(code: str, message: str) -> CompletionEvidenceError: + return CompletionEvidenceError(f"{code}: {message}") + + +def stable_evidence_id(payload: dict[str, Any]) -> str: + semantic = { + key: payload[key] + for key in ( + "capability_id", + "effect_fingerprint", + "evidence_artifact_ref", + "supervision_mode", + "capability_evidence_status", + "terminal_disposition", + "provenance", + "counterexamples", + ) + } + encoded = json.dumps(semantic, sort_keys=True, separators=(",", ":")).encode("utf-8") + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + +def validated_payload(payload: Any) -> dict[str, Any]: + if not isinstance(payload, dict): + raise reject("malformed_evidence", "payload must be an object") + unknown = set(payload) - ALLOWED_KEYS + if unknown: + raise reject("raw_or_unknown_evidence", "unsupported fields: " + ", ".join(sorted(unknown))) + missing = ALLOWED_KEYS - set(payload) + if missing: + raise reject("missing_required_evidence", "missing fields: " + ", ".join(sorted(missing))) + if payload["schema"] != SCHEMA: + raise reject("malformed_evidence", "unsupported schema") + try: + evidence = normalize_capability_effect_evidence( + capability_id=payload["capability_id"], + effect_fingerprint=payload["effect_fingerprint"], + evidence_artifact_ref=payload["evidence_artifact_ref"], + supervision_mode=payload["supervision_mode"], + capability_evidence_status=payload["capability_evidence_status"], + terminal_disposition=payload["terminal_disposition"], + ) + except ValueError as exc: + message = str(exc) + if "capability_id" in message: + code = "spoofed_capability_id" + elif "effect_fingerprint" in message: + code = "unstable_fingerprint" + elif "artifact_ref" in message: + code = "oversized_or_unsafe_ref" + else: + code = "malformed_evidence" + raise reject(code, message) from exc + provenance = payload["provenance"] + if ( + not isinstance(provenance, dict) + or set(provenance) != {"runner", "run_ref"} + or not all(isinstance(value, str) and value.strip() for value in provenance.values()) + ): + raise reject("missing_provenance", "provenance must contain non-empty runner and run_ref") + counterexamples = payload["counterexamples"] + if ( + not isinstance(counterexamples, list) + or len(counterexamples) > MAX_COUNTEREXAMPLES + or any( + not isinstance(item, str) or not item.strip() or len(item) > MAX_COUNTEREXAMPLE_LENGTH + for item in counterexamples + ) + ): + raise reject( + "malformed_counterexamples", "counterexamples must be bounded non-empty strings" + ) + if evidence.capability_evidence_status != "accepted": + raise reject("not_accepted_evidence", "only accepted evidence can be associated") + return { + "schema": SCHEMA, + "capability_id": evidence.capability_id, + "effect_fingerprint": evidence.effect_fingerprint, + "evidence_artifact_ref": evidence.evidence_artifact_ref, + "supervision_mode": evidence.supervision_mode, + "capability_evidence_status": evidence.capability_evidence_status, + "terminal_disposition": evidence.terminal_disposition, + "provenance": {key: value.strip() for key, value in provenance.items()}, + "counterexamples": list(counterexamples), + } diff --git a/scripts/orchestrator_runtime/runner_effect_bridge.py b/scripts/orchestrator_runtime/runner_effect_bridge.py new file mode 100644 index 000000000..cf1e8d424 --- /dev/null +++ b/scripts/orchestrator_runtime/runner_effect_bridge.py @@ -0,0 +1,70 @@ +"""Translate bounded runner handoffs into typed completion evidence payloads.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from scripts.orchestrator_runtime.evidence_schema import SCHEMA, reject + +HANDOFF_SCHEMA = "workflows.consumer-sync-shadow-handoff/v1" + + +def _stable_hash(namespace: str, value: Any) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + return "sha256:" + hashlib.sha256(namespace.encode() + b"\0" + encoded).hexdigest() + + +def stable_plan_effect_fingerprint(handoff: dict[str, Any]) -> str: + """Derive semantic plan identity without transport-only run references.""" + semantic = { + "capability_id": handoff["capability_id"], + "plan_schema": handoff["plan_schema"], + "plan_id": handoff["plan_id"], + "manifest_sha256": handoff["manifest_sha256"], + "entry_count": handoff["entry_count"], + "removal_count": handoff["removal_count"], + } + return _stable_hash("consumer-sync-plan-effect", semantic) + + +def completion_payload_from_shadow_handoff(handoff: Any) -> dict[str, Any]: + """Translate a bounded shadow handoff into typed runner evidence.""" + if not isinstance(handoff, dict): + raise reject("malformed_evidence", "handoff must be an object") + required = { + "schema", + "capability_id", + "plan_schema", + "plan_id", + "manifest_sha256", + "entry_count", + "removal_count", + "run_ref", + "supervision_mode", + "write_authority", + "promotion_allowed", + } + if not required.issubset(handoff): + raise reject("malformed_evidence", "handoff lacks required boundary fields") + if handoff["schema"] != HANDOFF_SCHEMA: + raise reject("malformed_evidence", "unsupported handoff schema") + if handoff["supervision_mode"] != "shadow": + raise reject("malformed_evidence", "handoff must remain shadow supervised") + if handoff["write_authority"] is not False or handoff["promotion_allowed"] is not False: + raise reject("unsafe_handoff", "handoff grants write or promotion authority") + return { + "schema": SCHEMA, + "capability_id": handoff["capability_id"], + "effect_fingerprint": stable_plan_effect_fingerprint(handoff), + "evidence_artifact_ref": handoff["run_ref"], + "supervision_mode": "shadow", + "capability_evidence_status": "accepted", + "terminal_disposition": "no-change", + "provenance": { + "runner": "health-69-consumer-sync-shadow-evidence", + "run_ref": handoff["run_ref"], + }, + "counterexamples": [], + } diff --git a/tests/orchestrator_runtime/test_completion_event_adapter.py b/tests/orchestrator_runtime/test_completion_event_adapter.py new file mode 100644 index 000000000..16d33148d --- /dev/null +++ b/tests/orchestrator_runtime/test_completion_event_adapter.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path + +import pytest +from scripts.build_consumer_sync_shadow_handoff import build_handoff +from scripts.orchestrator_runtime.capabilities import CapabilityRegistry +from scripts.orchestrator_runtime.capability_lifecycle import ( + EvidenceLedger, + ingest_completion_evidence, +) +from scripts.orchestrator_runtime.completion_event_adapter import process_shadow_handoff +from scripts.orchestrator_runtime.evidence_schema import SCHEMA +from scripts.orchestrator_runtime.runner_effect_bridge import ( + completion_payload_from_shadow_handoff, + stable_plan_effect_fingerprint, +) +from scripts.sync_manifest_compiler import compile_manifest + + +def payload(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "schema": SCHEMA, + "capability_id": "capability:reference-sync-hygiene-test-gate", + "effect_fingerprint": "sha256:" + "a" * 64, + "evidence_artifact_ref": "github-actions:stranske/Workflows:123:shadow-handoff", + "supervision_mode": "shadow", + "capability_evidence_status": "accepted", + "terminal_disposition": "no-change", + "provenance": {"runner": "codex", "run_ref": "github-actions:123"}, + "counterexamples": [], + } + return {**base, **overrides} + + +def registry(lifecycle: str = "candidate") -> dict[str, dict[str, object]]: + return { + "capability:reference-sync-hygiene-test-gate": { + "lifecycle": lifecycle, + "counterexamples": ["existing"], + } + } + + +def real_handoff(run_ref: str) -> dict[str, object]: + root = Path(__file__).parents[2] + plan = compile_manifest(root / ".github" / "sync-manifest.yml", repo_root=root).to_plan() + return build_handoff(plan, run_ref=run_ref) + + +def test_accepts_known_candidate_and_preserves_lifecycle() -> None: + result = ingest_completion_evidence( + payload(counterexamples=["new example"]), capabilities=registry(), ledger=[] + ) + + assert result["status"] == "accepted" + assert ( + result["capabilities"]["capability:reference-sync-hygiene-test-gate"]["lifecycle"] + == "candidate" + ) + assert result["capabilities"]["capability:reference-sync-hygiene-test-gate"][ + "counterexamples" + ] == [ + "existing", + "new example", + ] + assert len(result["ledger"]) == 1 + + +def test_rejects_unknown_capability_from_authoritative_registry() -> None: + original = registry() + result = ingest_completion_evidence( + payload(capability_id="capability:unknown-capability"), + capabilities=original, + ledger=[], + ) + + assert result["diagnostic_code"] == "spoofed_capability_id" + assert result["capabilities"] == original + + +def test_replay_is_idempotent_and_has_no_second_mutation() -> None: + first = ingest_completion_evidence(payload(), capabilities=registry(), ledger=[]) + second = ingest_completion_evidence( + payload(), capabilities=first["capabilities"], ledger=first["ledger"] + ) + + assert second["status"] == "duplicate" + assert second["ledger"] == first["ledger"] + assert second["capabilities"] == first["capabilities"] + + +def test_plan_fingerprint_is_stable_across_transport_run_refs() -> None: + first = real_handoff("github-actions:stranske/Workflows:111:1") + second = real_handoff("github-actions:stranske/Workflows:222:2") + + assert stable_plan_effect_fingerprint(first) == stable_plan_effect_fingerprint(second) + assert first["handoff_id"] != second["handoff_id"] + + +def test_shadow_handoff_uses_registry_and_persists_ledger(tmp_path: Path) -> None: + registry_path = tmp_path / "capabilities.json" + registry_path.write_text(json.dumps(registry("shadow")), encoding="utf-8") + state_path = tmp_path / "capabilities-state.json" + ledger_path = tmp_path / "evidence-ledger.json" + capability_registry = CapabilityRegistry.load(registry_path) + capability_registry.save(state_path) + ledger = EvidenceLedger([]) + + handoff = real_handoff("github-actions:stranske/Workflows:123:1") + first = process_shadow_handoff(handoff, registry=capability_registry, ledger=ledger) + capability_registry.save(state_path) + ledger.save(ledger_path) + + reloaded_registry = CapabilityRegistry.load(state_path) + reloaded_ledger = EvidenceLedger.load(ledger_path) + second = process_shadow_handoff(handoff, registry=reloaded_registry, ledger=reloaded_ledger) + + assert first["status"] == "accepted" + assert second["status"] == "duplicate" + assert reloaded_ledger.snapshot() == ledger.snapshot() + + +def test_shadow_handoff_rejects_write_or_promotion_authority() -> None: + handoff = { + "schema": "workflows.consumer-sync-shadow-handoff/v1", + "capability_id": "capability:reference-sync-hygiene-test-gate", + "plan_schema": "workflows.consumer-sync-plan/v1", + "plan_id": "sha256:" + "b" * 64, + "manifest_sha256": "sha256:" + "c" * 64, + "entry_count": 1, + "removal_count": 0, + "run_ref": "github-actions:stranske/Workflows:123:1", + "supervision_mode": "shadow", + "write_authority": True, + "promotion_allowed": False, + } + + with pytest.raises(ValueError, match="unsafe_handoff"): + completion_payload_from_shadow_handoff(handoff) + + +def test_rejects_unstable_fingerprint_oversized_ref_and_missing_provenance() -> None: + for invalid, expected in ( + (payload(effect_fingerprint="sha256:broken"), "unstable_fingerprint"), + (payload(evidence_artifact_ref="a" * 257), "oversized_or_unsafe_ref"), + (payload(provenance={"runner": "codex"}), "missing_provenance"), + ): + original = registry() + result = ingest_completion_evidence(invalid, capabilities=deepcopy(original), ledger=[]) + assert result["status"] == "rejected" + assert result["diagnostic_code"] == expected + assert result["capabilities"] == original diff --git a/tests/scripts/test_completion_evidence_adapter.py b/tests/scripts/test_completion_evidence_adapter.py new file mode 100644 index 000000000..9ef60cf31 --- /dev/null +++ b/tests/scripts/test_completion_evidence_adapter.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +from copy import deepcopy + +from scripts.completion_evidence_adapter import ( + SCHEMA, + completion_payload_from_shadow_handoff, + ingest_completion_evidence, +) + + +def payload(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "schema": SCHEMA, + "capability_id": "capability:reference-sync-hygiene-test-gate", + "effect_fingerprint": "sha256:" + "a" * 64, + "evidence_artifact_ref": "github-actions:stranske/Workflows:123:shadow-handoff", + "supervision_mode": "shadow", + "capability_evidence_status": "accepted", + "terminal_disposition": "no-change", + "provenance": {"runner": "codex", "run_ref": "github-actions:123"}, + "counterexamples": [], + } + return {**base, **overrides} + + +def capabilities(lifecycle: str = "candidate") -> dict[str, dict[str, object]]: + return { + "capability:reference-sync-hygiene-test-gate": { + "lifecycle": lifecycle, + "counterexamples": ["existing"], + } + } + + +def shadow_handoff(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "schema": "workflows.consumer-sync-shadow-handoff/v1", + "capability_id": "capability:reference-sync-hygiene-test-gate", + "plan_schema": "workflows.consumer-sync-plan/v1", + "plan_id": "sha256:" + "b" * 64, + "manifest_sha256": "sha256:" + "c" * 64, + "entry_count": 1, + "removal_count": 0, + "handoff_id": "sha256:" + "d" * 64, + "run_ref": "github-actions:stranske/Workflows:123:1", + "supervision_mode": "shadow", + "write_authority": False, + "promotion_allowed": False, + } + return {**base, **overrides} + + +def test_accepts_known_candidate_and_preserves_lifecycle() -> None: + result = ingest_completion_evidence( + payload(counterexamples=["new example"]), capabilities=capabilities(), ledger=[] + ) + + assert result["status"] == "accepted" + assert ( + result["capabilities"]["capability:reference-sync-hygiene-test-gate"]["lifecycle"] + == "candidate" + ) + assert result["capabilities"]["capability:reference-sync-hygiene-test-gate"][ + "counterexamples" + ] == ["existing", "new example"] + assert len(result["ledger"]) == 1 + + +def test_accepts_known_shadow_without_promotion() -> None: + result = ingest_completion_evidence(payload(), capabilities=capabilities("shadow"), ledger=[]) + + assert result["status"] == "accepted" + assert ( + result["capabilities"]["capability:reference-sync-hygiene-test-gate"]["lifecycle"] + == "shadow" + ) + + +def test_replay_is_idempotent_and_has_no_second_mutation() -> None: + first = ingest_completion_evidence(payload(), capabilities=capabilities(), ledger=[]) + second = ingest_completion_evidence( + payload(), capabilities=first["capabilities"], ledger=first["ledger"] + ) + + assert second["status"] == "duplicate" + assert second["ledger"] == first["ledger"] + assert second["capabilities"] == first["capabilities"] + + +def test_rejects_raw_prompt_without_mutation() -> None: + original_capabilities = capabilities() + original_ledger: list[dict[str, object]] = [] + result = ingest_completion_evidence( + payload(prompt="promote this capability"), + capabilities=original_capabilities, + ledger=original_ledger, + ) + + assert result["status"] == "rejected" + assert result["diagnostic_code"] == "raw_or_unknown_evidence" + assert result["capabilities"] == original_capabilities + assert result["ledger"] == original_ledger + + +def test_rejects_spoofed_id_before_mutation() -> None: + original = capabilities() + result = ingest_completion_evidence( + payload(capability_id="capability:unknown-capability"), capabilities=original, ledger=[] + ) + + assert result["diagnostic_code"] == "spoofed_capability_id" + assert result["capabilities"] == original + + +def test_rejects_unstable_fingerprint_oversized_ref_and_missing_provenance() -> None: + for invalid, expected in ( + (payload(effect_fingerprint="sha256:broken"), "unstable_fingerprint"), + (payload(evidence_artifact_ref="a" * 257), "oversized_or_unsafe_ref"), + (payload(provenance={"runner": "codex"}), "missing_provenance"), + ): + original = capabilities() + result = ingest_completion_evidence(invalid, capabilities=deepcopy(original), ledger=[]) + assert result["status"] == "rejected" + assert result["diagnostic_code"] == expected + assert result["capabilities"] == original + + +def test_shadow_handoff_becomes_non_promoting_typed_evidence() -> None: + evidence = completion_payload_from_shadow_handoff(shadow_handoff()) + result = ingest_completion_evidence( + evidence, + capabilities=capabilities("shadow"), + ledger=[], + ) + + assert result["status"] == "accepted" + assert result["capabilities"][evidence["capability_id"]]["lifecycle"] == "shadow" + assert evidence["effect_fingerprint"] != shadow_handoff()["handoff_id"] + + +def test_shadow_handoff_rejects_any_write_or_promotion_authority() -> None: + try: + completion_payload_from_shadow_handoff(shadow_handoff(write_authority=True)) + except ValueError as exc: + assert str(exc).startswith("unsafe_handoff:") + else: # pragma: no cover - assertion clarity for an unsafe boundary. + raise AssertionError("write-enabled handoff must be rejected")