diff --git a/docs/execution-plan.md b/docs/execution-plan.md index 46aa7c7..062068e 100644 --- a/docs/execution-plan.md +++ b/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 diff --git a/docs/orca-execution-plan.md b/docs/orca-execution-plan.md new file mode 100644 index 0000000..4e3cbe2 --- /dev/null +++ b/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. diff --git a/scripts/cycle_cli.py b/scripts/cycle_cli.py index 70113d1..8ef5e81 100644 --- a/scripts/cycle_cli.py +++ b/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)) diff --git a/scripts/cycle_dispatch.py b/scripts/cycle_dispatch.py index 78b5990..692ed3c 100644 --- a/scripts/cycle_dispatch.py +++ b/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 diff --git a/scripts/cycle_plan.py b/scripts/cycle_plan.py index 320f613..8104682 100644 --- a/scripts/cycle_plan.py +++ b/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: diff --git a/scripts/cycle_session.py b/scripts/cycle_session.py index a735cc3..7a06149 100644 --- a/scripts/cycle_session.py +++ b/scripts/cycle_session.py @@ -2,6 +2,9 @@ from __future__ import annotations from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from .cycle_journal import events as journal_events + from .plan_service import reconcile_plan + from .orca_plan import reconcile_outcome from .run_cycle import CycleHistoryStore, CycleInterrupted, CycleReport, Reported, RunStatuses from . import verification_plan as verification_contract @@ -15,6 +18,9 @@ from .workspace_setup import commit_setup, own_file from .cycle_plan import PlanRuntime else: + from cycle_journal import events as journal_events + from plan_service import reconcile_plan + from orca_plan import reconcile_outcome from cycle_state import SessionState from cycle_dispatch import StageDispatch from cycle_publication import PublicationHistory @@ -28,6 +34,8 @@ def __init__(self, api: Any, options: dict[str, Any]): # The facade supplies dependencies so existing integrations keep # the same public names and injection points. self.api = api + self.orca_context = options.get("orca_context") + self.orca_action = options.get("orca_action", "launch") self.state = SessionState() self.control = CycleControl() self.repo_id = options["repo_id"] @@ -307,8 +315,37 @@ def interrupted(self, interruption: CycleInterrupted) -> CycleReport: "was left as it was", stop_reason="interrupted") - def execute(self): + def execute(self) -> Any: try: + if self.orca_action != "launch": + # Resume/cancellation belong to the existing coordinator. + # Reconcile durable obligations before any attempt to launch; + # no documented live capability is assumed by this driver. + try: + record = self.plan_store.latest_plan( + self.repo_id, self.task_id, self.state.change_request_id) + if record is None or self.orca_context is None: + raise ValueError("Orca resume needs an existing context and persisted plan") + launches = [row["correlation"] for row in journal_events( + self.telemetry.path, self.repo_id, task_id=self.task_id, + kinds=("stage_result",)) if row.get("component") == "orca_launch"] + launch = launches[-1] if launches else None + if (launch is None + or launch.get("run_id") != self.orca_context.run_id + or launch.get("task_id") != self.orca_context.task_id + or launch.get("plan_id") != record["plan_id"]): + raise ValueError("Orca resume does not match the persisted launch identities") + self.execution_plan = reconcile_plan(record, self.authorized_bindings(), + head=self.plan_head) + self.state.verification_plan = self.execution_plan.verification + self.plan_conclusion = self.execution_plan.metadata.get("verification_conclusion") + self.save_plan("recovered") + outcome = reconcile_outcome(self.orca_context, self.execution_plan, + cancel=self.orca_action == "cancel") + return self.stop(outcome["reason"], "BLOCKED", + stop_reason="stage_not_completed") + except (ValueError, self.api.HistoryError) as error: + return self.stop(str(error), "BLOCKED", stop_reason="readiness_unconfirmed") try: self.api.journal_cleanup(self.telemetry.path, self.repo_id, (lambda request: self.checks_provider.pull(self.repo_id, request).state) diff --git a/scripts/executors.py b/scripts/executors.py index f906841..c18fcfe 100644 --- a/scripts/executors.py +++ b/scripts/executors.py @@ -47,7 +47,7 @@ from enum import Enum from pathlib import Path from urllib.parse import urlsplit -from typing import Any +from typing import Any, TYPE_CHECKING import stage_permissions as sp try: @@ -3157,6 +3157,21 @@ class OrcaDispatchContext: task_id: str worker_agent: str | None = None review_workspace: OrcaReviewWorkspace | None = None + execution_plan: Any = None + stage: str | None = None + task_spec_reader: Any = None + task_spec_installer: Any = None + + def task_spec(self) -> dict[str, Any]: + if self.execution_plan is None or self.stage is None: + raise ExecutorError("an Orca plan delegation needs an execution plan and stage") + if TYPE_CHECKING: + from .plan_service import delegation_spec + else: + from plan_service import delegation_spec + return {**delegation_spec(self.execution_plan, self.stage), + "run_id": self.run_id, "task_id": self.task_id, + "coordinator": self.coordinator} def __post_init__(self) -> None: for field_name in ("coordinator", "run_id", "task_id"): @@ -3223,6 +3238,18 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, "already exist; create them deliberately rather than having the " "dispatcher spawn them in someone's workspace", ) + # Legacy direct adapter callers remain launch-only. A managed cycle + # always supplies a plan and cannot dispatch an unselected stage. + task_spec = None + if context.execution_plan is not None: + try: + task_spec = context.task_spec() + if not callable(context.task_spec_reader): + return self._blocked(target, "orchestration_context", + "no verified Orca Task-spec read interface; the coordinator must " + "confirm the effective plan is installed in the existing Task") + except (ValueError, OSError, subprocess.SubprocessError, ExecutorError) as exc: + return self._blocked(target, "orchestration_context", str(exc)) review_workspace = None if not writes: review_workspace = context.review_workspace @@ -3252,6 +3279,30 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, f"no Orca agent is known for provider {target.provider!r}; " "name one in the dispatch context rather than guessing", ) + if task_spec is not None: + try: + bound = context.task_spec_reader(run_id=context.run_id, task_id=context.task_id) + if bound != task_spec and callable(context.task_spec_installer): + bound_plan = bound.get("plan") if isinstance(bound, dict) else None + if (not isinstance(bound, dict) or not isinstance(bound_plan, dict) + or any(bound.get(key) != task_spec[key] for key in + ("run_id", "task_id", "coordinator", "stage")) + or any(bound_plan.get(key) != task_spec["plan"][key] for key in + ("repository", "work_item"))): + return self._blocked(target, "orchestration_context", + "existing Orca Task binding is incomplete or belongs to another delegation") + # The native driver finalizes its revision just before + # dispatch. Refresh only this delegation, including Tasks + # initialized by the helper under a different plan ID. + # A writer's acknowledgement never replaces readback. + context.task_spec_installer(run_id=context.run_id, task_id=context.task_id, + spec=json.loads(json.dumps(task_spec))) + bound = context.task_spec_reader(run_id=context.run_id, task_id=context.task_id) + if bound != task_spec: + return self._blocked(target, "orchestration_context", + "existing Orca Task spec does not match the effective delegation plan") + except (ValueError, OSError, subprocess.SubprocessError, ExecutorError) as exc: + return self._blocked(target, "orchestration_context", str(exc)) # `task` here is the Task ID Orca already holds, not the prompt. The # prompt went into that Task when it was created; passing prose to # --task would silently create work nobody can find again. @@ -3318,6 +3369,9 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, state = result.get("state") artifacts = {"argv": argv, "dispatchId": result.get("dispatchId"), "state": state, "launch": result.get("launch")} + if task_spec is not None: + artifacts["task_spec"] = task_spec + artifacts["functional_state"] = "result_pending" start_state = "started" if result.get("dispatchId") else "unknown" if review_workspace is not None: artifacts["reviewWorkspace"] = { @@ -3447,6 +3501,8 @@ def dispatch( registry = registry or Registry() target = decision.target adapter = registry.get(target.executor) + if not isinstance(adapter, OrcaAdapter): + kw.pop("context", None) on_progress = kw.pop("on_progress", None) on_workspace = kw.pop("on_workspace", None) dispatch_attempt_id = kw.pop("dispatch_attempt_id", None) @@ -3564,6 +3620,10 @@ def dispatch( ) policy_note = None + if isinstance(adapter, OrcaAdapter): + # Consumed by the shared gates above, not Orca worker-start options. + kw.pop("publishes", None) + kw.pop("publication_permissions", None) if command_policy is not None and isinstance(adapter, OrcaAdapter): # The table and hook are applied to native dispatches only (#142). # Orca workers keep their behaviour and are not presented as diff --git a/scripts/fast-path-tests.json b/scripts/fast-path-tests.json index 687c7b8..ecfe80d 100644 --- a/scripts/fast-path-tests.json +++ b/scripts/fast-path-tests.json @@ -1,5 +1,9 @@ { "groups": { + "orca-plan-documentation": { + "paths": ["docs/orca-execution-plan.md"], + "tests": ["test_orca_plan.OrcaPlanTests.test_modes_and_live_evidence_obligation_are_documented"] + }, "repository-configuration": { "paths": [".code-cycle.yml"], "tests": [ diff --git a/scripts/orca_plan.py b/scripts/orca_plan.py new file mode 100644 index 0000000..f516d5f --- /dev/null +++ b/scripts/orca_plan.py @@ -0,0 +1,138 @@ +"""Installed plan helper for the existing Orca coordinator, not a coordinator. + +No Orca CLI or undocumented endpoint is invoked here. Run/Task ownership and +the source of authoritative outcomes remain with the installed coordinator. +""" +from __future__ import annotations + +import argparse +import json +import re +import sqlite3 +from dataclasses import replace +from pathlib import Path +from typing import Any, TYPE_CHECKING + +from cycle_history import CycleHistoryStore +from execution_plan import ContractError, ExecutionPlan +if TYPE_CHECKING: + from .plan_service import (delegation_spec, persist_plan, receipt_for, + reconcile_plan, require_security) +else: + from plan_service import (delegation_spec, persist_plan, receipt_for, + reconcile_plan, require_security) +from telemetry import Telemetry + + +def reconcile_outcome(context, plan, *, coordinator=None, cancel=False) -> dict[str, Any]: + """Consume a capability supplied by the existing coordinator. + + The stock CLI has no verified binding for this capability. Tests inject a + coordinator, never convert a launch receipt or worker_done to an outcome. + """ + capability = "request_cancellation" if cancel else "functional_outcome" + if (context.execution_plan is not None + and receipt_for(context.execution_plan) != receipt_for(plan)): + return {"status": "BLOCKED", "reason": "delegation plan was invalidated; reconcile its evidence", + **receipt_for(plan)} + method = getattr(coordinator, capability, None) + if not callable(method): + return {"status": "BLOCKED", "capability": capability, + "reason": "no verified authoritative Orca coordinator interface", + **receipt_for(plan)} + if not isinstance(plan.head_sha, str) or not re.fullmatch(r"[0-9a-f]{40}", plan.head_sha): + return {"status": "BLOCKED", "reason": "authoritative outcome needs the applicable full HEAD", + **receipt_for(plan)} + result = method(run_id=context.run_id, task_id=context.task_id) + if cancel: + # Cancellation acknowledgement is not functional completion either. + if (not isinstance(result, dict) or result.get("accepted") is not True + or result.get("run_id") != context.run_id + or result.get("task_id") != context.task_id): + return {"status": "BLOCKED", "reason": "coordinator did not confirm the cancellation request", + **receipt_for(plan)} + return {"status": "RESULT_PENDING", "cancellation_requested": True, + **receipt_for(plan)} + if result is None: + return {"status": "RESULT_PENDING", **receipt_for(plan)} + expected = {**receipt_for(plan), "run_id": context.run_id, + "task_id": context.task_id, "head_sha": plan.head_sha, + "stage": context.stage, "repo": plan.repository} + if (not isinstance(result, dict) or type(result.get("schema")) is not int + or result.get("schema") != 2 or type(result.get("revision")) is not int + or any(result.get(key) != value for key, value in expected.items()) + or result.get("evidence_valid") is not True + or result.get("terminal") is not True): + return {"status": "BLOCKED", "reason": "stale, partial or mismatched coordinator outcome", + **receipt_for(plan)} + from execution_plan import stage_definition + accepted = stage_definition(context.stage).completes_with | {"BLOCKED", "FAILED", "CANCELLED"} + if (context.stage in {"review", "rereview"} + and plan.control("ci").timing == "parallel"): + accepted = accepted | {"REVIEW_PASSED_AWAITING_CI"} + if result.get("status") not in accepted: + return {"status": "BLOCKED", "reason": "no terminal functional outcome", + **receipt_for(plan)} + # This only recovers an outcome. Existing stage/check validators and exit + # gates must still judge its functional payload, findings and evidence. + return result + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=("initialize", "reconcile", "spec", "security", "rereview")) + parser.add_argument("--database", required=True) + parser.add_argument("--repo", required=True) + parser.add_argument("--task", required=True, help="work-item ID, not Orca Task ID") + parser.add_argument("--cycle", required=True) + parser.add_argument("--pr") + parser.add_argument("--plan", help="ExecutionPlan JSON from run_cycle --dry-run --json") + parser.add_argument("--head", required=True) + parser.add_argument("--command", action="append", default=[], help="currently authorized binding") + parser.add_argument("--stage") + parser.add_argument("--changes-after-review", action="store_true") + parser.add_argument("--open-findings", action="store_true") + args = parser.parse_args(argv) + try: + if not re.fullmatch(r"[0-9a-f]{40}", args.head): + raise ContractError("--head needs the full current commit SHA") + # Telemetry owns the schema. This never changes a project database. + telemetry = Telemetry(Path(args.database)) + store = CycleHistoryStore(telemetry.path) + record = store.latest_plan(args.repo, args.task, args.pr) + if args.action == "initialize": + if record is not None: + raise ContractError("a persisted plan exists; reconcile it instead of replacing it") + if not args.plan: + raise ContractError("initialize needs --plan from the shared resolver") + plan = ExecutionPlan.from_dict(json.loads(Path(args.plan).read_text(encoding="utf-8"))) + if plan.repository != args.repo or plan.work_item != args.task: + raise ContractError("plan scope does not match the requested work item") + plan = replace(plan, head_sha=args.head) + else: + if record is None: + raise ContractError("no recoverable shared plan") + plan = reconcile_plan(record, args.command, head=args.head, + changes_after_review=args.changes_after_review if args.action == "rereview" else None, + open_findings=args.open_findings) + if args.action == "security": + plan = require_security(plan, "Orca coordinator observed security gate") + if args.action == "spec": + assert record is not None + if plan.head_sha != record["head_sha"]: + raise ContractError("HEAD changed; reconcile and persist before delegation") + output = delegation_spec(plan, args.stage) + else: + plan, receipt = persist_plan(plan, store, cycle_id=args.cycle, + change_request_id=args.pr, + provenance="initial" if args.action == "initialize" else "recovered") + output = {**receipt, "plan": plan.to_dict()} + print(json.dumps(output, sort_keys=True)) + return 0 + except (OSError, ValueError, KeyError, TypeError, sqlite3.Error) as exc: + print(json.dumps({"status": "BLOCKED", "reason": str(exc)})) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/plan_service.py b/scripts/plan_service.py new file mode 100644 index 0000000..9e1d067 --- /dev/null +++ b/scripts/plan_service.py @@ -0,0 +1,85 @@ +"""Transport-independent plan operations; no recorder, worker or driver state.""" +from __future__ import annotations + +from dataclasses import replace +from typing import Any + +from cycle_history import CycleHistoryStore, plan_digest, plan_projection, restore_plan +from execution_plan import ExecutionPlan, ExecutionState, Obligation, Resolution, StagePolicy +from execution_profiles import apply_rereview_decision, rereview_decision + + +def receipt_for(plan: ExecutionPlan) -> dict[str, Any]: + plan.validate() + return {"plan_id": plan.plan_id, "revision": plan.revision, + "digest": plan_digest(plan_projection(plan))} + + +def persist_plan(plan: ExecutionPlan, store: CycleHistoryStore, *, cycle_id: str, + change_request_id: str | None = None, provenance: str = "observed", + stop_reason: str | None = None) -> tuple[ExecutionPlan, dict[str, Any]]: + """Append a revision through the same privacy-safe store for every transport.""" + plan = replace(plan, revision=plan.revision + 1) + plan.validate() + receipt = store.save_plan(plan, cycle_id=cycle_id, + change_request_id=change_request_id, provenance=provenance, stop_reason=stop_reason) + return plan, receipt + + +def invalidate_head(plan: ExecutionPlan, head: str | None) -> ExecutionPlan: + """Keep obligations and historical checks, invalidate only HEAD-bound evidence.""" + if head == plan.head_sha: + return plan + return replace(plan, head_sha=head, stages=tuple( + replace(item, state=ExecutionState.PLANNED, conclusion=None) + if item.stage_id in {"verify", "security"} and item.selected else item + for item in plan.stages), metadata={**plan.metadata, + "verification_conclusion": "not_verified"}) + + +def require_security(plan: ExecutionPlan, reason: str) -> ExecutionPlan: + security = plan.stage("security") + if security.resolution is Resolution.PENDING: + return plan.resolve_decision("security", True, reason, "runtime") + if security.policy is StagePolicy.DISABLED: + 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) + + +def reconcile_plan(record: dict[str, Any], commands: tuple[str, ...] | list[str], *, + head: str | None, changes_after_review: bool | None = None, + open_findings: bool = False) -> ExecutionPlan: + """Recover canonical identity and obligations before deciding any new work. + + Commands are rebound only from today's authorization. Partial results never + erase checks, and verification results remain tied to their original HEAD. + Review invalidation must be observed by the coordinator, not guessed here. + """ + plan = invalidate_head(restore_plan(record, tuple(commands)), head) + if changes_after_review is not None: + decision = rereview_decision(changes_after_review=changes_after_review, + open_findings=open_findings) + plan = apply_rereview_decision(plan, decision) + plan.validate() + return plan + + +def delegation_spec(plan: ExecutionPlan, role: str) -> dict[str, Any]: + """Data the coordinator puts in its existing Task before worker-start.""" + plan.check_executable(role) + stage = plan.stage(role) + from execution_plan import stage_definition + definition = stage_definition(role) + return {**receipt_for(plan), "objective": plan.objective, + "execution_profile": plan.execution_profile, "stage": role, + "head_sha": plan.head_sha, "provenance": stage.provenance, + "workspace_policy": definition.workspace_policy.value, + "publication_permissions": list(definition.publication_permissions), + "security_required": plan.stage("security").selected, + "verify_selected": plan.stage("verify").selected, + "plan": plan.to_dict()} diff --git a/scripts/run_cycle.py b/scripts/run_cycle.py index ef06289..2fb8a4f 100644 --- a/scripts/run_cycle.py +++ b/scripts/run_cycle.py @@ -58,7 +58,7 @@ from dataclasses import dataclass, field, replace from datetime import datetime, timezone from pathlib import Path, PureWindowsPath -from typing import Any, TYPE_CHECKING +from typing import Any, TYPE_CHECKING, cast if TYPE_CHECKING: from . import verification_plan as verification_contract @@ -2361,6 +2361,8 @@ def run_cycle( criteria_reader: Callable[..., verification_contract.Criteria] | None = None, execution_plan: ExecutionPlan | None = None, security_rule=None, + orca_context=None, + orca_action: str = "launch", ) -> CycleReport: """[issue_review ->] implement -> review -> (resolve -> rereview)*, recorded. @@ -2417,7 +2419,9 @@ def run_cycle( reported, or `publication_unverified` when that state cannot be read. """ from cycle_session import CycleSession - return CycleSession(sys.modules[__name__], locals()).execute() + # The facade supplies this module's types; the session's static aliases + # may otherwise resolve through the package-qualified module name. + return cast(CycleReport, cast(object, CycleSession(sys.modules[__name__], locals()).execute())) def _started_elsewhere(outcome: StageOutcome) -> bool: """Whether this stage launched work that finishes outside this process.""" diff --git a/scripts/runtime.manifest b/scripts/runtime.manifest index e92ad7f..b4a81bb 100644 --- a/scripts/runtime.manifest +++ b/scripts/runtime.manifest @@ -61,3 +61,5 @@ cycle_workspace.py workspace_setup.py cycle_state.py cycle_plan.py +plan_service.py +orca_plan.py diff --git a/scripts/windows-test-shards.json b/scripts/windows-test-shards.json index a66d5b9..c1179f8 100644 --- a/scripts/windows-test-shards.json +++ b/scripts/windows-test-shards.json @@ -53,6 +53,7 @@ "test_host_permissions", "test_manual_rereview", "test_native_prerequisites", + "test_orca_plan", "test_permission_view", "test_permission_lifecycle", "test_publication", @@ -101,6 +102,7 @@ "test_execution_plan", "test_execution_profiles", "test_verification_plan", + "test_orca_plan", "test_stats", "test_support", "test_validate_package" diff --git a/skills/cc-orca-orchestrator/SKILL.md b/skills/cc-orca-orchestrator/SKILL.md index 2f3c953..6f6bc54 100644 --- a/skills/cc-orca-orchestrator/SKILL.md +++ b/skills/cc-orca-orchestrator/SKILL.md @@ -165,6 +165,63 @@ identifier stays as written. A stop that needs no decision carries no ## Responsibilities +### Shared execution plan + +Use the installed `run_cycle.py --dry-run --json` resolver with the repository +configuration and explicit `execution_profile`, `objective` and stage overrides. +Do not implement an Orca activation matrix. All five profiles (Prototype, +Standard, Full, Hotfix, Custom) consume the same `ExecutionPlan`. See +`docs/orca-execution-plan.md` for examples and terminal obligations. + +Initialize that plan with the installed `orca_plan.py initialize` helper and +the coordinator's telemetry database, cycle ID, repository, work item and +current HEAD. Its `reconcile`, `security`, `rereview` and `spec` actions use the +shared planner operations, validators and `CycleHistoryStore.save_plan`. +Supply only currently authorized command bindings with `--command`; persisted +fingerprints never authorize execution. An existing plan is reconciled, never +replaced by initialization. Preserve plan ID, revision history, canonical +digest, provenance, frozen checks and required controls. + +Before each Task, reconcile current HEAD and outstanding obligations and obtain +the helper's `spec` for its stage. Put the complete response in the existing +Task spec, including effective profile, plan receipt, stage, HEAD, verification +and security requirements, workspace and publication permissions. Give the +worker the matching `plan_id`, `revision`, `digest` and schema-2 receipt +requirements. Construct `OrcaDispatchContext` from existing Run, Task and +coordinator identities and that effective plan; do not invent identities or +grant permissions from worker output. Keep the original review-workspace rules. +Supply a verified `task_spec_reader` binding from the existing coordinator when +using the adapter; it must read back the actual existing Task's complete bound +spec. Without that capability the adapter blocks before launch. An expected +spec in an artifact is not evidence that the Task received it. + +Before advancing, obtain the authoritative functional outcome from the existing +coordinator and validate Run/Task identity, repository, stage, applicable HEAD, +plan receipt, evidence validity and terminal status. A launch receipt, +`worker_done`, a matching schema-2 receipt alone, or a partial result cannot +complete a stage. Run the existing functional-result, finding, frozen-check, +security and CI validators. A helper spec or recovery response is no approval. + +On resume reconcile persisted plan revisions, receipts, HEAD and outstanding +checks before dispatch. Invalidate HEAD-bound verification/security completion +without erasing old evidence or obligations. Unsupported authoritative-outcome, +resume or cancellation interfaces in the installed Orca version stop `BLOCKED`; +never invent a protocol or launch a duplicate worker to recover a result. +Cancellation stays owned by the existing coordinator. The native driver's +`--orca-action resume|cancel` fails closed until a live binding is verified. + +After review, call helper action `rereview` with observed relevant changes or +invalidated review evidence (`--changes-after-review`) and open findings +(`--open-findings`). Rereview is conditional in every profile, including Full. +An unchanged clean review stands; unchanged open findings keep their blocks. +Never start an identical review in search of another result. Changed evidence +reactivates only the controls it invalidates. A planned stage is dispatched +only after its shared decision selects it; required controls cannot be removed. + +Record the installed Orca version, verified capabilities and limitations with +the evidence. The exact-HEAD host campaign `orca-live` is required and stays +unverified until it passes; fixtures and launch success do not satisfy it. + Do: - run `cc-provider-bootstrap` before creating the Orca Run; @@ -794,9 +851,12 @@ Apply the table in *Issue review before implementation*: only a confirmed ### From `cc-resolve-comments` -- `RESOLVED`: continue to a `cc-rereview` Task. +- `RESOLVED`: reconcile the shared rereview decision against relevant changes + and invalidated evidence. Dispatch `cc-rereview` only when selected; retain + any unchanged open finding's blocks without repeating the review. - `PARTIALLY_RESOLVED`: continue to `cc-rereview` only when the next rereview - can verify what remains, that is when `blocking` is `false` and no entry in + can verify what remains, the shared plan selects rereview after a relevant + change or invalidated evidence, `blocking` is `false` and no entry in `unresolved_findings` states that external intervention, access, a decision, or an unavailable environment is required. Otherwise stop with `HUMAN_INTERVENTION` and list the unresolved findings. diff --git a/tests/test_orca_plan.py b/tests/test_orca_plan.py new file mode 100644 index 0000000..9bafc9f --- /dev/null +++ b/tests/test_orca_plan.py @@ -0,0 +1,412 @@ +"""Shared policy parity and fail-closed asynchronous Orca boundaries (#330).""" +from __future__ import annotations + +import contextlib +import io +import json +import subprocess +import sys +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +import executors as ex +import orca_plan +from test_run_cycle import RunCycleTestCase, Talker +import router +from cycle_journal import events +from cycle_history import CycleHistoryStore +from execution_plan import ExecutionState, ContractError +from execution_profiles import (ExecutionSettings, build_plan, invocation_settings, + resolve_execution, apply_rereview_decision, rereview_decision) +from plan_service import (delegation_spec, invalidate_head, persist_plan, + receipt_for, reconcile_plan, require_security) +from telemetry import Telemetry +from verification_plan import Check, CheckResult, VerificationPlan, parse_criteria + +HEAD = "a" * 40 +NEXT = "b" * 40 +MODES = ("prototype", "standard", "full", "hotfix", "custom") + + +def sample(mode="standard"): + return replace(build_plan(resolve_execution(ExecutionSettings(), + invocation_settings(execution_profile=mode)), + repository="datacas/code-cycle-toolkit", work_item="330", + plan_id="plan-test", provenance="run"), head_sha=HEAD) + + +class OrcaPlanTests(unittest.TestCase): + def test_frozen_check_recovery_preserves_partial_history(self): + command = "python3 -m unittest" + verification = VerificationPlan(parse_criteria("github", "## Acceptance criteria\n- Works")) + verification = verification.extend((Check("check-1", command, "codex", "deep_coder", + "project", True, ("AC-1",), provenance="runtime", reason="runtime"),)) + verification = verification.record(CheckResult("check-1", HEAD, "pass", + ({"exit_code": 0, "executed": 1},))) + with tempfile.TemporaryDirectory() as directory: + store = CycleHistoryStore(Telemetry(Path(directory) / "db.sqlite").path) + plan, _ = persist_plan(replace(sample(), verification=verification), store, cycle_id="cycle") + record = store.latest_plan(plan.repository, plan.work_item, None) + with self.assertRaisesRegex(ValueError, "authorized command"): + reconcile_plan(record, (), head=HEAD) + current = reconcile_plan(record, (command,), head=HEAD) + self.assertEqual(current.verification.current("check-1", HEAD), "pass") + moved = reconcile_plan(record, (command,), head=NEXT) + self.assertEqual(moved.verification.current("check-1", NEXT), "unverified") + self.assertEqual(moved.verification.results, verification.results) + self.assertEqual(moved.verification.checks, verification.checks) + + def test_equivalent_inputs_and_task_specs_in_every_mode(self): + for mode in MODES: + with self.subTest(mode=mode): + native = sample(mode) + context = ex.OrcaDispatchContext("coordinator", "run", "task", + execution_plan=native, stage="implement") + spec = context.task_spec() + self.assertEqual(spec["plan"], native.to_dict()) + self.assertEqual(spec["execution_profile"], mode) + self.assertEqual({key: spec[key] for key in receipt_for(native)}, receipt_for(native)) + self.assertEqual(spec["publication_permissions"], ["comment", "create_pr", "push_branch"]) + + def test_rereview_is_conditional_in_every_mode_and_retains_blocks(self): + for mode in MODES: + for changes, findings in ((False, False), (False, True), (True, True)): + with self.subTest(mode=mode, changes=changes, findings=findings): + decision = rereview_decision(changes_after_review=changes, open_findings=findings) + original = sample(mode) + if mode == "prototype": + original = replace(build_plan(resolve_execution(ExecutionSettings(), + invocation_settings(execution_profile=mode, objective="merge_ready", + stages=("review=required",))), repository="datacas/code-cycle-toolkit", + work_item="330", plan_id="plan-test"), head_sha=HEAD) + plan = apply_rereview_decision(original, decision) + self.assertEqual(plan.stage("rereview").selected, changes) + self.assertEqual(decision.blocks_retained, findings and not changes) + + def test_head_invalidation_retains_required_security_and_old_evidence(self): + plan = require_security(sample(), "gate") + plan = replace(plan, stages=tuple(replace(stage, state=ExecutionState.COMPLETED, + conclusion="completed") if stage.stage_id == "security" else stage for stage in plan.stages)) + changed = invalidate_head(plan, NEXT) + self.assertEqual(changed.stage("security").state, ExecutionState.PLANNED) + self.assertTrue(changed.stage("security").selected) + self.assertEqual(changed.stage("implement"), plan.stage("implement")) + self.assertEqual(changed.verification, plan.verification) + self.assertIs(invalidate_head(plan, HEAD), plan) + + def test_canonical_history_and_reconcile_preserve_identity(self): + with tempfile.TemporaryDirectory() as directory: + store = CycleHistoryStore(Telemetry(Path(directory) / "db.sqlite").path) + plan, receipt = persist_plan(sample("full"), store, cycle_id="cycle") + self.assertEqual(receipt, receipt_for(plan)) + record = store.latest_plan(plan.repository, plan.work_item, None) + restored = reconcile_plan(record, (), head=HEAD, + changes_after_review=False, open_findings=True) + self.assertEqual(restored.plan_id, plan.plan_id) + self.assertGreater(restored.revision, plan.revision) + self.assertFalse(restored.stage("rereview").selected) + moved = reconcile_plan(record, (), head=NEXT) + self.assertEqual(moved.head_sha, NEXT) + + def test_authoritative_outcome_requires_all_bindings(self): + plan = sample() + context = ex.OrcaDispatchContext("coordinator", "run", "task", execution_plan=plan, stage="implement") + valid = {"schema": 2, **receipt_for(plan), "run_id": "run", "task_id": "task", + "repo": plan.repository, "stage": "implement", "head_sha": HEAD, + "terminal": True, "evidence_valid": True, "status": "IMPLEMENTED"} + coordinator = Mock() + coordinator.functional_outcome.return_value = valid + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator), valid) + for key, value in (("run_id", "other"), ("task_id", "other"), ("revision", 90), + ("digest", "x"), ("head_sha", NEXT), ("terminal", False), + ("evidence_valid", False), ("status", "worker_done"), ("schema", 1)): + with self.subTest(key=key): + coordinator.functional_outcome.return_value = {**valid, key: value} + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator)["status"], "BLOCKED") + coordinator.functional_outcome.return_value = {"ok": True, "result": {"state": "ready"}} + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator)["status"], "BLOCKED") + coordinator.functional_outcome.return_value = None + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator)["status"], "RESULT_PENDING") + self.assertEqual(orca_plan.reconcile_outcome(context, plan)["status"], "BLOCKED") + self.assertEqual(orca_plan.reconcile_outcome(context, plan, cancel=True)["status"], "BLOCKED") + coordinator.request_cancellation.return_value = {"accepted": True, "run_id": "run", "task_id": "task"} + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator, cancel=True)["status"], "RESULT_PENDING") + coordinator.request_cancellation.assert_called_once_with(run_id="run", task_id="task") + coordinator.request_cancellation.return_value = {"accepted": False} + self.assertEqual(orca_plan.reconcile_outcome(context, plan, coordinator=coordinator, cancel=True)["status"], "BLOCKED") + + def test_launch_is_only_pending_and_review_workspace_is_preserved(self): + target = SimpleNamespace(provider="openai", model="gpt-test", effort="medium") + plan = sample() + context = ex.OrcaDispatchContext("coordinator", "run", "task", execution_plan=plan, stage="implement") + runner = Mock(return_value=subprocess.CompletedProcess([], 0, json.dumps({"ok": True, + "result": {"dispatchId": "dispatch", "state": "ready"}}), "")) + adapter = ex.OrcaAdapter() + unsupported = adapter.dispatch(target, "ignored", cwd="/tmp/source", context=context, + writes=True, runner=runner) + self.assertEqual(unsupported.outcome, ex.DispatchOutcome.BLOCKED) + runner.assert_not_called() + mismatch = replace(context, task_spec_reader=lambda **_: {"revision": 99}) + self.assertEqual(adapter.dispatch(target, "ignored", context=mismatch, + writes=True, runner=runner).outcome, ex.DispatchOutcome.BLOCKED) + runner.assert_not_called() + installed = json.loads(json.dumps(context.task_spec())) + context = replace(context, task_spec_reader=lambda **_: installed) + result = adapter.dispatch(target, "ignored", cwd="/tmp/source", context=context, + writes=True, runner=runner) + self.assertTrue(result.asynchronous) + self.assertEqual(result.artifacts["functional_state"], "result_pending") + self.assertEqual(result.artifacts["task_spec"]["revision"], plan.revision) + self.assertFalse(result.agent_output) + review = replace(context, stage="review") + installed_review = json.loads(json.dumps(review.task_spec())) + review = replace(review, task_spec_reader=lambda **_: installed_review) + refused = adapter.dispatch(target, "ignored", cwd="/tmp/source", context=review, runner=runner) + self.assertEqual(refused.missing_capability, "review_workspace_isolation") + review = replace(review, review_workspace=ex.OrcaReviewWorkspace( + "/tmp/review", "/tmp/source", "disposable")) + launched = adapter.dispatch(target, "ignored", cwd="/tmp/source", context=review, runner=runner) + self.assertTrue(launched.asynchronous) + self.assertEqual(launched.artifacts["reviewWorkspace"]["path"], "/tmp/review") + self.assertEqual(launched.artifacts["task_spec"]["publication_permissions"], ["comment"]) + + def test_installation_requires_independent_matching_readback_and_preconditions(self): + target = router.Target("orca", "openai", "gpt-test", "medium") + context = ex.OrcaDispatchContext("coordinator", "run", "task", + execution_plan=replace(sample(), revision=2), stage="implement") + runner = Mock(return_value=subprocess.CompletedProcess([], 0, json.dumps({"ok": True, + "result": {"dispatchId": "dispatch", "state": "ready"}}), "")) + adapter = ex.OrcaAdapter() + installer = Mock(return_value=True) + # An acknowledgement or an installer without a reader cannot launch. + missing = replace(context, task_spec_installer=installer) + self.assertEqual(adapter.dispatch(target, "ignored", context=missing, + writes=True, runner=runner).outcome, ex.DispatchOutcome.BLOCKED) + installer.assert_not_called() + stale = replace(context, execution_plan=sample()).task_spec() + reader = Mock(return_value=stale) + pending = replace(missing, task_spec_reader=reader) + self.assertEqual(adapter.dispatch(target, "ignored", context=pending, + writes=True, runner=runner).outcome, ex.DispatchOutcome.BLOCKED) + self.assertEqual(reader.call_count, 2) + installer.assert_called_once_with(run_id="run", task_id="task", spec=context.task_spec()) + runner.assert_not_called() + # A failed workspace gate must not update the existing Task. + installer.reset_mock() + self.assertEqual(adapter.dispatch(target, "ignored", context=pending, + runner=runner).missing_capability, "review_workspace_isolation") + installer.assert_not_called() + # A callback cannot mutate the expected spec to make stale readback pass. + def corrupt(*, spec, **_): + spec.clear() + spec["revision"] = 0 + self.assertEqual(adapter.dispatch(target, "ignored", context=replace(pending, + task_spec_installer=corrupt), writes=True, runner=runner).outcome, + ex.DispatchOutcome.BLOCKED) + # Failed installation remains blocked, and does not start a worker. + installer.side_effect = OSError("Task installation failed") + result = adapter.dispatch(target, "ignored", context=pending, writes=True, runner=runner) + self.assertEqual(result.outcome, ex.DispatchOutcome.BLOCKED) + self.assertIn("Task installation failed", result.detail) + runner.assert_not_called() + # A Task already bound to the final plan needs no write capability. + reader.return_value = json.loads(json.dumps(context.task_spec())) + installer.reset_mock() + self.assertTrue(adapter.dispatch(target, "ignored", context=pending, + writes=True, runner=runner).asynchronous) + installer.assert_not_called() + + def test_installer_rejects_foreign_and_incomplete_task_bindings(self): + target = router.Target("orca", "openai", "gpt-test", "medium") + context = ex.OrcaDispatchContext("coordinator", "run", "task", + execution_plan=sample(), stage="implement") + spec = context.task_spec() + foreign = [{**spec, key: "other"} for key in + ("run_id", "task_id", "coordinator")] + foreign.append({**spec, "stage": "review"}) + foreign.extend({**spec, "plan": {**spec["plan"], key: "other"}} + for key in ("repository", "work_item")) + incomplete = [None, {}, [], False, {"revision": 0}, + {key: value for key, value in spec.items() if key != "run_id"}, + {**spec, "plan": None}, {**spec, "plan": {}}, + {**spec, "plan": {"repository": spec["plan"]["repository"]}}] + for index, bound in enumerate([*foreign, *incomplete]): + with self.subTest(case=index): + stored = {"spec": bound} + runner = Mock(return_value=subprocess.CompletedProcess([], 0, + json.dumps({"ok": True, "result": {"dispatchId": "dispatch", + "state": "ready"}}), "")) + installer = Mock(side_effect=lambda **kw: stored.update(spec=kw["spec"])) + reader = Mock(side_effect=lambda **_: stored["spec"]) + result = ex.OrcaAdapter().dispatch(target, "ignored", writes=True, + context=replace(context, task_spec_reader=reader, + task_spec_installer=installer), runner=runner) + self.assertEqual(result.outcome, ex.DispatchOutcome.BLOCKED) + installer.assert_not_called() + runner.assert_not_called() + reader.assert_called_once_with(run_id="run", task_id="task") + + def test_installer_updates_only_the_same_delegation_across_plan_ids(self): + target = router.Target("orca", "openai", "gpt-test", "medium") + context = ex.OrcaDispatchContext("coordinator", "run", "task", + execution_plan=sample(), stage="implement") + # A helper-created Task predates the native driver's finalized plan. + original = replace(context, execution_plan=replace(sample(), + plan_id="run:datacas/code-cycle-toolkit#330")).task_spec() + stored = json.loads(json.dumps(original)) + def install(*, run_id, task_id, spec): + nonlocal stored + self.assertEqual((run_id, task_id), ("run", "task")) + stored = json.loads(json.dumps(spec)) + installer = Mock(side_effect=install) + reader = Mock(side_effect=lambda **_: stored) + runner = Mock(return_value=subprocess.CompletedProcess([], 0, json.dumps({"ok": True, + "result": {"dispatchId": "dispatch", "state": "ready"}}), "")) + result = ex.OrcaAdapter().dispatch(target, "ignored", writes=True, + context=replace(context, task_spec_reader=reader, + task_spec_installer=installer), runner=runner) + self.assertTrue(result.asynchronous) + self.assertEqual(result.artifacts["functional_state"], "result_pending") + self.assertEqual(stored, context.task_spec()) + self.assertNotEqual(original["plan_id"], stored["plan_id"]) + installer.assert_called_once() + self.assertEqual(reader.call_count, 2) + runner.assert_called_once() + + def test_helper_boundary_and_installed_manifest(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) + (path / "plan.json").write_text(json.dumps(sample().to_dict())) + base = [sys.executable, str(ROOT / "scripts/orca_plan.py")] + args = ["--database", str(path / "db.sqlite"), "--repo", "datacas/code-cycle-toolkit", + "--task", "330", "--cycle", "cycle", "--head", HEAD] + result = subprocess.run([*base, "initialize", *args, "--plan", str(path / "plan.json")], + capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + initial = json.loads(result.stdout) + spec = subprocess.run([*base, "spec", *args, "--stage", "implement"], capture_output=True, text=True) + self.assertEqual(spec.returncode, 0, spec.stderr) + self.assertEqual(json.loads(spec.stdout)["digest"], initial["digest"]) + duplicate = subprocess.run([*base, "initialize", *args, "--plan", str(path / "plan.json")], + capture_output=True, text=True) + self.assertEqual(duplicate.returncode, 1) + manifest = (ROOT / "scripts/runtime.manifest").read_text() + self.assertIn("orca_plan.py", manifest) + self.assertIn("plan_service.py", manifest) + + def test_modes_and_live_evidence_obligation_are_documented(self): + document = (ROOT / "docs/orca-execution-plan.md").read_text() + for mode in MODES: + self.assertIn(f"--execution-profile {mode}", document) + for token in ("orca-live", "unverified", "commit SHA", "worker_done", "1.4.209"): + self.assertIn(token, document) + + +class FixtureOrca(ex.OrcaAdapter): + requires_publication_preflight = False + + def __init__(self, *, bound=True, install=True): + super().__init__() + self.contexts = [] + self.bound = bound + self.install = install + self.installs = [] + # An existing Task installed before the native driver selects a stage. + self.stored_spec = ex.OrcaDispatchContext("coordinator", "run", "task", + execution_plan=replace(sample(), repository="owner/api", work_item="API-7", + plan_id="run:owner/api#API-7"), stage="implement").task_spec() + + def install_spec(self, *, run_id, task_id, spec): + self.installs.append((run_id, task_id, spec)) + self.stored_spec = json.loads(json.dumps(spec)) + + def probe(self): + return ex.ProbeResult(self.name, ex.Availability.READY, "fixture") + + def publication_access(self, probe, *, writes): + return True, "fixture" + + def dispatch(self, target, task, **kw): + context = kw.get("context") + if context is not None and self.bound: + # Fixture of an existing coordinator reading its actual Task spec. + context = replace(context, task_spec_reader=lambda **_: self.stored_spec, + task_spec_installer=self.install_spec if self.install else None) + kw["context"] = context + self.contexts.append(kw.get("context")) + return super().dispatch(target, task, **kw, runner=lambda *args, **kwargs: + subprocess.CompletedProcess([], 0, json.dumps({"ok": True, + "result": {"dispatchId": "fixture-dispatch", "state": "ready"}}), "")) + + +class OrcaDriverTests(RunCycleTestCase): + def launch(self, *, bound=True, install=True, **kw): + adapter = FixtureOrca(bound=bound, install=install) + profiles = router.load_profiles({"code_cycle": {"profiles": { + "cheap_coder": {"primary": "orca:openai/gpt-test medium"}, + "deep_coder": {"primary": "orca:openai/gpt-test medium"}}}}) + report = self.run_cycle(adapter, Talker("claude"), profiles=profiles, + orca_context=ex.OrcaDispatchContext("coordinator", "run", "task"), **kw) + return report, adapter + + def test_driver_constructs_context_and_persists_pending_launch(self): + report, adapter = self.launch() + self.assertEqual(report.status, "IN_PROGRESS", report.stopped_because) + self.assertEqual(len(adapter.contexts), 1) + context = adapter.contexts[0] + self.assertEqual(context.stage, "implement") + self.assertEqual(context.execution_plan.repository, "owner/api") + self.assertEqual(context.execution_plan.work_item, "API-7") + self.assertEqual(len(adapter.installs), 1) + self.assertEqual(adapter.installs[0][:2], ("run", "task")) + self.assertEqual(adapter.stored_spec, context.task_spec()) + self.assertGreater(adapter.stored_spec["revision"], 0) + launches = [row for row in events(self.store.path, "owner/api", task_id="API-7") + if row["component"] == "orca_launch"] + self.assertEqual(len(launches), 1) + self.assertEqual(launches[0]["correlation"]["dispatch_id"], "fixture-dispatch") + self.assertEqual(launches[0]["correlation"]["revision"], str(adapter.stored_spec["revision"])) + self.assertEqual(launches[0]["correlation"]["digest"], adapter.stored_spec["digest"]) + + def test_preinstalled_task_without_coordinator_installer_is_blocked(self): + report, adapter = self.launch(install=False) + self.assertEqual(report.status, "BLOCKED") + self.assertIn("Task spec does not match", report.stopped_because or "") + self.assertEqual(adapter.stored_spec["revision"], 1) + self.assertEqual(adapter.stored_spec["plan_id"], "run:owner/api#API-7") + self.assertFalse(adapter.installs) + + def test_resume_and_cancellation_do_not_relaunch_or_accept_partial_results(self): + self.launch() + for action in ("resume", "cancel"): + report, adapter = self.launch(orca_action=action) + self.assertEqual(report.status, "BLOCKED") + self.assertIn("no verified authoritative", report.stopped_because) + self.assertFalse(adapter.contexts) + + def test_missing_history_blocks_without_a_launch(self): + report, adapter = self.launch(orca_action="resume") + self.assertEqual(report.status, "BLOCKED") + self.assertFalse(adapter.contexts) + + def test_missing_task_binding_is_blocked_in_the_driver(self): + report, _ = self.launch(bound=False) + self.assertEqual(report.status, "BLOCKED") + self.assertIn("no verified Orca Task-spec", report.stopped_because) + + def test_resume_rejects_a_different_existing_task(self): + self.launch() + report = self.run_cycle(FixtureOrca(), Talker("claude"), orca_action="resume", + orca_context=ex.OrcaDispatchContext("coordinator", "run", "other-task")) + self.assertEqual(report.status, "BLOCKED") + self.assertIn("persisted launch identities", report.stopped_because) + + +if __name__ == "__main__": + unittest.main()