feat(orca): share execution plans in supervised cycles (#330)

docs/execution-plan.md

@@ -1,5 +1,9 @@
 # Stage registry and execution-plan contract
 
+Supervised Orca routes use the same plan operations and receipt format. See
+[Orca execution plans](orca-execution-plan.md) for helper commands, the five
+modes, asynchronous outcomes and the required live verification evidence.
+
 The runtime keeps stage descriptions in `scripts/execution_plan.py`. A stage
 names its role, responsibility, output contract, workspace policy, publication
 permissions, skill or function, and accepted completion statuses. Runtime-owned

docs/orca-execution-plan.md

@@ -0,0 +1,123 @@
+# Shared execution plans in supervised Orca cycles
+
+Orca owns Runs, Tasks, workers, deliveries, cancellation and terminal release.
+The toolkit owns policy: `resolve_execution` and `build_plan` resolve every
+route. `plan_service.py` shares canonical receipts, persistence, HEAD
+invalidation, security obligations and rereview reconciliation with the native
+driver. There is no second coordinator.
+
+Resolve configuration and invocation overrides with the installed runtime.
+Save the resulting JSON outside the checkout. These examples use placeholders
+for the coordinator's real cycle identifier, current HEAD and authorized commands.
+
+```bash
+python3 ~/.code-cycle/runtime/run_cycle.py --repo owner/repo --task 330 --execution-profile prototype --dry-run --json
+python3 ~/.code-cycle/runtime/run_cycle.py --repo owner/repo --task 330 --execution-profile standard --dry-run --json
+python3 ~/.code-cycle/runtime/run_cycle.py --repo owner/repo --task 330 --execution-profile full --dry-run --json
+python3 ~/.code-cycle/runtime/run_cycle.py --repo owner/repo --task 330 --execution-profile hotfix --dry-run --json
+python3 ~/.code-cycle/runtime/run_cycle.py --repo owner/repo --task 330 --execution-profile custom --stage security=required --dry-run --json
+python3 ~/.code-cycle/runtime/orca_plan.py initialize --database /tmp/orca-plan.sqlite --repo owner/repo --task 330 --cycle existing-cycle --head CURRENT_SHA --plan /tmp/plan.json
+python3 ~/.code-cycle/runtime/orca_plan.py spec --database /tmp/orca-plan.sqlite --repo owner/repo --task 330 --cycle existing-cycle --head CURRENT_SHA --stage implement
+```
+
+`initialize` refuses to replace an existing plan. `reconcile` restores the
+latest scoped revision with `CycleHistoryStore`, rebinds command fingerprints
+only from current `--command` arguments, and persists a new revision. History
+grants no execution permission. Missing bindings block recovery. Reconcile
+HEAD before `spec`: exporting an unpersisted HEAD is refused. Writes append
+revision history, preserving canonical plan ID, provenance, digest format and
+obligations. The database is coordinator telemetry, not an application database.
+
+Before creating its existing Task, the coordinator embeds the entire `spec`
+response: effective profile, plan, stage, receipt, HEAD, security/verification
+requirements, workspace policy and publication permissions. It supplies the
+existing Run, Task and coordinator identities to `OrcaDispatchContext`. The
+adapter's `task_spec()` exposes the same data and launch artifacts retain it.
+Since `worker-start` consumes a pre-existing Task, the coordinator must install
+this spec first; an artifact alone cannot prove the actual Task contents.
+The managed adapter requires a coordinator-supplied `task_spec_reader` that
+reads that existing Task and returns its bound spec. It compares the complete
+spec before launching. Missing or mismatched bindings block before worker-start.
+For a native driver launch the plan is finalized and persisted during stage
+selection, after the coordinator created its Task. Supply an explicit
+`task_spec_installer(run_id, task_id, spec)` capability alongside the reader to
+install that final spec in the **existing** Task when readback differs and its
+binding already matches this delegation's `run_id`, `task_id`, `coordinator`,
+`stage`, `plan.repository` and `plan.work_item`. A foreign or incomplete binding,
+including an empty binding, blocks before installation or worker-start. The
+coordinator must bind these identities when it creates the Task. A stale plan
+revision or contents may be refreshed within that binding; `plan_id` alone is
+not an ownership key, since the helper and native driver can use different plan
+IDs for the same delegation. The
+installer must be supplied by the existing coordinator; it does not create Tasks
+or define a new Orca CLI command. The adapter checks workspace and provider
+preconditions first, passes a detached copy of the final spec to the installer,
+and reads the Task again. Installation acknowledgement alone never authorizes
+worker-start: the independent reader must return the complete matching spec.
+Without an installer, a preinstalled matching spec still works; a stale one
+blocks. An installer without a reader also blocks.
+The stock CLI has no verified Task-spec reader binding and therefore blocks
+planned Orca launches until that integration is supplied; API fixtures exercise
+the successful path without inventing a command in the installed Orca CLI.
+
+The native driver accepts `--orca-coordinator`, `--orca-run` and `--orca-task`
+together; `--task` remains the provider work item. Stages routed to Orca receive
+the effective plan and stage in their context. The driver records the launch
+dispatch ID and receipt in its journal and stops at `IN_PROGRESS` with
+`result_pending` launch evidence. A launch
+receipt or `worker_done` alone is never a functional result. Review still needs
+an explicit isolated `OrcaReviewWorkspace`, and publication permissions retain
+their existing gates. The native CLI does not create an Orca review workspace;
+an API caller or coordinator supplies its contract, otherwise review blocks.
+
+`--orca-action resume` and `--orca-action cancel` reconcile the persisted plan,
+HEAD and outstanding checks before doing anything else. The stock route then
+returns `BLOCKED` with `orca_capability_unsupported`: no authoritative-outcome
+or cancellation CLI binding has been verified. It launches no replacement
+worker. `orca_plan.reconcile_outcome` is the boundary for a capability supplied
+by the existing coordinator, not a remote protocol. A file or worker self-report
+cannot supply authority. Cancellation acknowledgement remains `RESULT_PENDING`.
+
+Recovered outcomes require schema 2, expected Run/Task/repository/stage, plan ID,
+revision, digest, applicable HEAD, terminal status and valid evidence. Partial,
+stale, launch-only and mismatched responses block. This boundary only recovers
+an outcome: existing functional-result, finding, verification and CI validators
+must still judge its payload before advancing. Unsupported interfaces block
+instead of inventing polling or cancellation commands.
+
+| Mode | Default terminal obligation | Rereview |
+|---|---|---|
+| Prototype | `local_change`; native driver stops `HUMAN_INTERVENTION` with `local_only`, without approval; publishing or approval adds guarantees | Conditional when approval is requested |
+| Standard | `READY_FOR_MANUAL_MERGE` after review, verification, applicable security and CI | Conditional |
+| Full | Same merge gates, with every applicable declared control | Conditional; never an identical repeat |
+| Hotfix | Same merge gates plus test-first reproduction and focused regression | Conditional |
+| Custom | Standard plus overrides; required obligations cannot be withdrawn | Conditional |
+
+All modes retain failures and partial evidence, stop `BLOCKED` on missing
+capabilities and keep merge manual. Transport `RESULT_PENDING` is never a
+functional terminal outcome. An explicit objective, including `verified_change`,
+is resolved through the shared plan and determines its own obligations.
+
+After review, helper action `rereview` accepts `--changes-after-review` only for
+observed relevant changes or invalidated review evidence. Without that flag,
+`--open-findings` retains blocks without another identical review. Full uses
+the same rule. HEAD changes invalidate current verification and security
+completion; old check results keep their original SHA. Other stages and
+required obligations remain. Action `security` adds the control when the shared
+security gate requires it and cannot remove it.
+
+## Required live evidence
+
+Required check `orca-live` stays **unverified** until a host campaign passes on
+the exact commit SHA. The implementer sandbox does not run Orca or weaken
+isolation. Orca **1.4.209** was reported at planning time, not observed by this
+implementation.
+
+The host verifier records actual version, environment, exact commands, commit
+SHA, Run/Task/worker IDs, plan receipt, actual Task contents, launch receipt and
+authoritative functional outcome. It exercises launch, resume, partial/stale
+results, cancellation and terminal release; checks supported interfaces against
+the installed version; and records limitations. Missing authoritative-outcome
+or cancellation interfaces are unsupported capabilities, never fabricated
+success. Fixtures complement this campaign and cannot satisfy `orca-live` or
+mark the issue complete.

scripts/cycle_cli.py

@@ -13,6 +13,10 @@ def main(api: Any, argv: list[str] | None = None) -> int:
     parser.add_argument("--repo", default=None,
                         help=("repository identifier, owner/name; defaults to "
                               f"code_cycle.repository.selector in {api.CONFIG_NAME}"))
+    parser.add_argument("--orca-coordinator")
+    parser.add_argument("--orca-run")
+    parser.add_argument("--orca-task", help="existing Orca Task ID, distinct from --task")
+    parser.add_argument("--orca-action", choices=("launch", "resume", "cancel"), default="launch")
     parser.add_argument("--task", required=True, help="work item identifier")
     parser.add_argument(
         "--work-item-repository", default=None,
@@ -105,6 +109,13 @@ def main(api: Any, argv: list[str] | None = None) -> int:
     parser.add_argument("--json", action="store_true",
                         help="with --dry-run, print the plan as its JSON contract")
     args = parser.parse_args(argv)
+    orca_context = None
+    orca_ids = (args.orca_coordinator, args.orca_run, args.orca_task)
+    if any(orca_ids) or args.orca_action != "launch":
+        if not all(orca_ids):
+            parser.error("Orca needs all of --orca-coordinator, --orca-run and --orca-task")
+        from executors import OrcaDispatchContext
+        orca_context = OrcaDispatchContext(*orca_ids)
     if args.progress_interval <= 0:
         parser.error("--progress-interval must be greater than zero")
     if args.json and not args.dry_run:
@@ -259,6 +270,8 @@ def main(api: Any, argv: list[str] | None = None) -> int:
                 consent=consent,
                 execution_plan=execution_plan,
                 security_rule=api.GateRule.from_config(config),
+                orca_context=orca_context,
+                orca_action=args.orca_action,
             )
     except (api.CycleDriverError, api.GrantError) as error:
         parser.error(str(error))

scripts/cycle_dispatch.py

@@ -55,6 +55,9 @@ def run(self: Any, role: str, instruction: str = "", *,
         stage_start_clean = self.cwd is not None and self.api._clean_worktree(self.cwd)
         evidence = ""
         stage_kwargs = self.dispatch_kwargs
+        if self.orca_context is not None:
+            stage_kwargs = {**stage_kwargs, "context": self.api.replace(
+                self.orca_context, execution_plan=self.execution_plan, stage=role)}
         artifact = None
         tampered = None
         # No base to measure against means no file, so no directory either.
@@ -75,7 +78,7 @@ def run(self: Any, role: str, instruction: str = "", *,
                             if directory is not None else None)
                 evidence = artifact.prompt() if artifact is not None else self.api.NO_DIFF_FILE
                 if artifact is not None:
-                    stage_kwargs = {**self.dispatch_kwargs, "read_dirs": (str(directory),)}
+                    stage_kwargs = {**stage_kwargs, "read_dirs": (str(directory),)}
             if role in {"implement", "resolve"} and self.allowed_test_command:
                 stage_kwargs = {
                     **stage_kwargs,
@@ -261,9 +264,21 @@ def unfinished(self: Any, outcome: StageOutcome, reported: Reported,
                             for reason in outcome.decision.reasons))
             )
             return self.stop(self.api._why(outcome), reported=reported,
-                        status="BLOCKED" if blocked_effort else self.api.UNRESOLVED_END,
+                        status="BLOCKED" if blocked_effort or (
+                            self.orca_context is not None and outcome.result is not None
+                            and outcome.result.executor == "orca"
+                            and outcome.result.outcome.value == "blocked") else self.api.UNRESOLVED_END,
                         stop_reason="dispatch_failed")
-        if self.api._started_elsewhere(outcome):
+        if self.api._started_elsewhere(outcome) and outcome.result is not None:
+            if self.orca_context is not None and outcome.result.executor == "orca":
+                self.journal.record("stage_result", component="orca_launch", status="result_pending",
+                    correlation={"run_id": self.orca_context.run_id,
+                        "task_id": self.orca_context.task_id,
+                        "head_sha": self.plan_head,
+                        "dispatch_id": outcome.result.artifacts.get("dispatchId"),
+                        **(self.plan_receipt or {})})
+                return self.stop(self.api._elsewhere(outcome), "IN_PROGRESS",
+                                 stop_reason="stage_not_completed")
             return self.stop(self.api._elsewhere(outcome), reported=reported)
         return None
 

scripts/cycle_plan.py

@@ -8,6 +8,7 @@
 from typing import Any, TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from .plan_service import invalidate_head, persist_plan, require_security
     from .cycle_journal import events as journal_events
     from .cycle_history import HistoryError, command_digest, restore_plan
     from .execution_plan import ContractError, ExecutionPlan, ExecutionState, Obligation, Resolution, StagePolicy
@@ -17,6 +18,7 @@
     from .verification_plan import (CheckResult, VerificationError, VerificationPlan,
                                     configured_check, is_manual_command)
 else:
+    from plan_service import invalidate_head, persist_plan, require_security
     from cycle_journal import events as journal_events
     from cycle_history import HistoryError, command_digest, restore_plan
     from execution_plan import ContractError, ExecutionPlan, ExecutionState, Obligation, Resolution, StagePolicy
@@ -113,13 +115,12 @@ def save_plan(self: Any, provenance: str = "observed", stop_reason=None) -> None
                 if item.stage_id == "verify" and item.selected else item
                 for item in self.execution_plan.stages))
         self.execution_plan = replace(self.execution_plan,
-            revision=self.execution_plan.revision + 1,
             verification=self.state.verification_plan,
             head_sha=self.plan_head,
             metadata={"verification_conclusion": self.plan_conclusion})
         self.execution_plan.validate()
-        self.plan_receipt = self.plan_store.save_plan(
-            self.execution_plan, cycle_id=self.recorder.cycle_id,
+        self.execution_plan, self.plan_receipt = persist_plan(
+            self.execution_plan, self.plan_store, cycle_id=self.recorder.cycle_id,
             change_request_id=self.state.change_request_id,
             provenance=provenance, stop_reason=stop_reason)
         self.report.execution_plan = self.plan_receipt
@@ -303,10 +304,7 @@ def select_plan_stage(self: Any, role: str) -> None:
         if observed is not None and observed != self.plan_head:
             self.plan_head = observed
             self.plan_conclusion = "not_verified"
-            self.execution_plan = replace(self.execution_plan, stages=tuple(
-                replace(item, state=ExecutionState.PLANNED, conclusion=None)
-                if item.stage_id in {"verify", "security"} and item.selected else item
-                for item in self.execution_plan.stages))
+            self.execution_plan = invalidate_head(self.execution_plan, observed)
         plan = self.execution_plan
         stage = plan.stage(role)
         if role == "resolve" and stage.resolution is Resolution.NOT_APPLICABLE:
@@ -342,18 +340,7 @@ def select_plan_stage(self: Any, role: str) -> None:
         self.save_plan()
 
     def require_security(self: Any, plan, reason: str):
-        security = plan.stage("security")
-        if security.resolution is Resolution.PENDING:
-            return plan.resolve_decision("security", True, reason, "runtime")
-        if security.policy is StagePolicy.DISABLED:
-            # Existing obligation validation explains the policy conflict.
-            return plan.settle_obligation("security_policy", True, reason)
-        plan = replace(plan, stages=tuple(
-            replace(item, selected=True, obligation=Obligation.REQUIRED,
-                    state=ExecutionState.PLANNED if not item.selected else item.state,
-                    resolution=Resolution.RESOLVED)
-            if item.stage_id == "security" else item for item in plan.stages))
-        return plan.settle_obligation("security_policy", True, reason)
+        return require_security(plan, reason)
 
     def record_plan_stage(self: Any, role: str, reported, outcome, test_payload=None) -> None:
         payload = reported.payload or {}
@@ -365,10 +352,7 @@ def record_plan_stage(self: Any, role: str, reported, outcome, test_payload=None
                     raise ContractError("reported evidence does not match the workspace HEAD")
             if self.plan_head and head != self.plan_head:
                 self.plan_conclusion = "not_verified"
-                self.execution_plan = replace(self.execution_plan, stages=tuple(
-                    replace(item, state=ExecutionState.PLANNED, conclusion=None)
-                    if item.stage_id in {"verify", "security"} and item.selected else item
-                    for item in self.execution_plan.stages))
+                self.execution_plan = invalidate_head(self.execution_plan, head)
             self.plan_head = head
         conclusion = self.api._legacy_tests(test_payload).get("verification")
         if conclusion is not None: