feat: record typed completion evidence for shadow sync

.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,11 +93,15 @@ 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']}")
           print(f"- Removals: {handoff['removal_count']}")
           print('- Supervision: shadow')
           print('- Write authority: false')
           print('- Promotion allowed: false')
+          print(f"- Completion evidence: {completion_evidence['status']}")
           PY

config/orchestrator_runtime/capabilities.json

@@ -0,0 +1,6 @@
+{
+  "capability:reference-sync-hygiene-test-gate": {
+    "counterexamples": [],
+    "lifecycle": "shadow"
+  }
+}

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).

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) |

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.

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",

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",
+]

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",
+]

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)

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,
+        }