diff --git a/doc/architecture/SPEC/app/core.md b/doc/architecture/SPEC/app/core.md index f9533fab..017b2e56 100644 --- a/doc/architecture/SPEC/app/core.md +++ b/doc/architecture/SPEC/app/core.md @@ -16,8 +16,11 @@ blocks the run. Execution rechecks the context after planning and refuses a changed checkout or policy before starting the executor. `Copilot` keeps reservation and run-path methods as compatibility delegates -to `app.reservation.RunReservation`. It owns workflow resolution and child -execution; it does not own idempotency indexing or run ID validation. +to `app.reservation.RunReservation`. It owns workflow resolution, confirmation, +reserved-run claims, and terminal status writes. `app.workflow_execution` +owns the executor lifecycle, locks, tracing, and terminal outcome; the old +`Copilot._execute` path delegates and retains the last-run summary. The +application does not own idempotency indexing or run ID validation. The pre-execution gate runs before a normal CLI run directory is created. The child rechecks the read-only policy and checkout binding. Existing diff --git a/doc/architecture/SPEC/app/workflow_execution.md b/doc/architecture/SPEC/app/workflow_execution.md new file mode 100644 index 00000000..82df37b7 --- /dev/null +++ b/doc/architecture/SPEC/app/workflow_execution.md @@ -0,0 +1,21 @@ +# app/workflow_execution.py — workflow execution boundary + + + +`WorkflowExecution` runs an already selected playbook with the caller's +current LLM dependency. It rechecks the repository context before starting, +owns the run and shared knowledge locks, +initializes tracing and notification, seeds executor state with checkout and +adapter policy, and drives `Executor` through `run_guarded`. It returns an +immutable `ExecutionResult` containing the exit code and blocked reason. + +A losing run lock leaves active run artifacts untouched. A failed knowledge +lock releases an acquired run lock. A caller-provided lock stays owned by the +caller. Finalizers and locks run on every executor outcome, including failure. +The application owns planning, confirmation, durable reservation, and the +reserved worker's terminal `run_status.json` write. CLI and MCP transports do +not belong in this module. + +`Copilot._execute` remains the compatibility entry point and records +`last_run_dir`/`last_blocked_reason` from the returned result. Existing run +state keys, traces, exit codes, and checkpoint behavior stay stable. diff --git a/src/infermatrix_copilot/app/core.py b/src/infermatrix_copilot/app/core.py index de3001dc..cc747bbc 100644 --- a/src/infermatrix_copilot/app/core.py +++ b/src/infermatrix_copilot/app/core.py @@ -2,13 +2,12 @@ generate) → plan-review gate → executor, plus the compound-command queue, resume, and the /status /logs /playbooks built-ins. -This is the orchestration core; the argparse/REPL wiring lives in `entry.py` and -metrics formatting is shared with `metrics.py`. +This is the orchestration core; the argparse/REPL wiring lives in `entry.py`, +and executor lifecycle lives in `app.workflow_execution`. """ from __future__ import annotations -import asyncio import json import os import time @@ -19,23 +18,19 @@ from .. import run_status as rs from ..adapters.base import AdapterError from ..config import Settings, TierNotConfiguredError -from ..engine.executor import Executor -from ..engine.lifecycle import RunLock, RunLockHeld, run_guarded +from ..engine.lifecycle import RunLock, RunLockHeld from ..engine.planner import Planner, PlanningError, Resolution from ..engine.registry import StepRegistry from ..engine.steps import register_builtin_steps -from ..llm import LLM -from ..notify import BLOCKED_EXIT, Notifier +from ..notify import BLOCKED_EXIT from ..playbooks.store import PlaybookStore, parse_playbook, playbook_to_doc -from ..push import PushPolicy from ..review.plan_gate import _mode_review_context, review_plan_gate from ..review.reviewer import run_plan_review -from ..run_trace import RunTrace from ..task_spec import TaskSpec from ..ui import style -from ..metrics import format_metrics_line from .reservation import RunReservation from .repository_context import RepositoryContext, RepositoryContextResolver +from .workflow_execution import WorkflowExecution class GateOutcome(NamedTuple): @@ -81,6 +76,9 @@ def __init__(self, settings: Settings | None = None): self.registry = register_builtin_steps(StepRegistry()) self.store = PlaybookStore(self.settings.playbooks_dir, self.registry) self.planner = Planner(self.store, self.registry) + self.workflow_execution = WorkflowExecution( + self.settings, self.registry, self.repository_context, + ) self.last_run_dir: Path | None = None # why the last execution stopped, when it stopped blocked — read by # `_execute_reserved_locked`, which owns the terminal run_status write @@ -267,117 +265,16 @@ def _execute(self, playbook, spec: TaskSpec, run_dir: Path, *, resolution_mode: str = "resume", tier: str = "?", resuming: bool = False, held_lock: RunLock | None = None, planned_context: RepositoryContext | None = None) -> int: - """Run a resolved `playbook` to completion in `run_dir`: init tracing + - notifier, seed the shared state (repo path, push policy, protected - branches / high-risk modules from the adapter when present), drive the - Executor, then print per-step marks, optional run metrics, and the final - status. `resuming` seeds the state so steps can pick up where they left - off. Returns the exit code (0 done, BLOCKED_EXIT blocked, else 1), and - records `last_blocked_reason` for the reserved-run caller, which has only - the code but owns the terminal `run_status.json` write.""" + """Compatibility entry point; application keeps the last-run summary.""" self.last_run_dir = run_dir self.last_blocked_reason = "" - try: - repo_context = self.repository_context.for_spec(spec) - except AdapterError as exc: - self.last_blocked_reason = f"repository context invalid: {exc}" - print(style("✋ ", "red", "bold") + self.last_blocked_reason) - return BLOCKED_EXIT - if planned_context is not None and repo_context != planned_context: - self.last_blocked_reason = "repository context changed after planning" - print(style("✋ ", "red", "bold") + self.last_blocked_reason) - return BLOCKED_EXIT - lock = held_lock - if lock is None: - try: - # First, before any trace/status write: a losing concurrent - # invocation (second --resume) must leave the active run's - # artifacts completely untouched. execute_reserved holds the - # lock across its whole lifecycle and passes it in instead. - lock = RunLock(run_dir).acquire() - except RunLockHeld as exc: - print(style("✋ ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - # Per-repo knowledge lock, SHARED, held for the run's lifetime: - # never contends with other runs; it exists so the knowledge - # migration's EXCLUSIVE acquire can prove no potential store - # writer is alive (and so no run starts mid-migration). - knowledge_lock = None - if spec.repo: - from ..memory.paths import (KnowledgeLockHeld, - KnowledgePaths, - KnowledgeRunLock, - KnowledgeStateError) - try: - knowledge_lock = KnowledgeRunLock( - KnowledgePaths.resolve(self.settings, spec.repo) - .knowledge_run_lock).acquire_shared() - except (KnowledgeLockHeld, KnowledgeStateError) as exc: - # BOTH refusals take the terminal protocol: an invalid - # activation marker (KnowledgeStateError) must exit - # blocked/3 with the run lock RELEASED, never escape as - # a traceback that leaves the lock held in a long-lived - # process (PR-boundary F4) - if held_lock is None: - lock.release() - print(style("✋ ", "red", "bold") + str(exc)) - return BLOCKED_EXIT - try: - from .. import tracing - tracing.init(run_dir.name, run_dir / "trace.jsonl") - # Stamp the workflow into the span file itself, so a trace lifted out of - # its run directory still says which playbook and task produced it. - tracing.run_meta(playbook=f"{playbook.name}@{playbook.version}", - task_kind=spec.kind, repo=spec.repo, tier=tier, - mode=spec.mode, resolution=resolution_mode, - report_only=spec.report_only, post=spec.post, - resuming=resuming, params=spec.params) - trace = RunTrace(run_dir / "run_trace.jsonl") - notifier = Notifier(self.settings, run_dir, trace, run_dir.name) - trace.record("task", spec=spec.model_dump(), resolution=resolution_mode, - playbook=playbook.name, tier=tier) - state: dict = { - "task_spec": spec.model_dump(), - "repo_path": repo_context.repo_path, - "push_policy": PushPolicy(), # steps may replace with a derived policy - "protected_branches": list(repo_context.protected_branches), - "resuming": resuming, - } - if repo_context.high_risk_modules: - state["high_risk_modules"] = list(repo_context.high_risk_modules) - executor = Executor(self.registry, self.settings, run_dir=run_dir, - trace=trace, llm=self.llm, notifier=notifier) - # run_guarded finalizes inside the event loop: playbooks that - # register run finalizers (lifecycle.register_finalizer) get - # teardown on every exit path. Nothing registered == no-op. - outcome = asyncio.run( - run_guarded(executor.run(playbook, state), run_dir)) - - if outcome.status == "done": - notifier.resolve() - - if self.settings.metrics_enabled: - try: # metrics are facts about the run; never let them break it - from ..metrics import collect_run_metrics - m = collect_run_metrics(run_dir, self.settings, outcome.status) - print(format_metrics_line(m, run_dir)) - except Exception as exc: - trace.record("metrics_error", error=f"{type(exc).__name__}: {exc}") - - for step_id, r in outcome.step_results.items(): - mark = style("✓", "green") if r.ok else style("✗", "red", "bold") - print(f" {mark} {step_id}: {r.summary}") - print(f"run {run_dir.name}: {outcome.status} ({run_dir})") - if outcome.status == "blocked": - self.last_blocked_reason = str(outcome.blocked_reason or "") - print(style(" ⚠ ", "yellow", "bold") + f"{outcome.blocked_reason}\n see {run_dir / 'ESCALATION.md'}") - return BLOCKED_EXIT - return 0 if outcome.status == "done" else 1 - finally: - if knowledge_lock is not None: - knowledge_lock.release() - if held_lock is None: - lock.release() + result = self.workflow_execution.run( + playbook, spec, run_dir, resolution_mode=resolution_mode, + tier=tier, resuming=resuming, held_lock=held_lock, + planned_context=planned_context, llm=self.llm, + ) + self.last_blocked_reason = result.blocked_reason + return result.exit_code def run_queue(self, specs: list[TaskSpec], *, assume_yes: bool = False, plan_only: bool = False) -> int: diff --git a/src/infermatrix_copilot/app/workflow_execution.py b/src/infermatrix_copilot/app/workflow_execution.py new file mode 100644 index 00000000..d38198da --- /dev/null +++ b/src/infermatrix_copilot/app/workflow_execution.py @@ -0,0 +1,154 @@ +"""Execute an already selected workflow with run and knowledge locks.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from pathlib import Path + +from ..adapters.base import AdapterError +from ..config import Settings +from ..engine.executor import Executor +from ..engine.lifecycle import RunLock, RunLockHeld, run_guarded +from ..engine.registry import StepRegistry +from ..metrics import format_metrics_line +from ..notify import BLOCKED_EXIT, Notifier +from ..push import PushPolicy +from ..run_trace import RunTrace +from ..task_spec import TaskSpec +from ..ui import style +from .repository_context import RepositoryContext, RepositoryContextResolver + + +@dataclass(frozen=True) +class ExecutionResult: + exit_code: int + blocked_reason: str = "" + + +class WorkflowExecution: + """Own the executor, locks, tracing and terminal outcome for one run.""" + + def __init__( + self, settings: Settings, registry: StepRegistry, + repository_context: RepositoryContextResolver, + ) -> None: + self.settings = settings + self.registry = registry + self.repository_context = repository_context + + def run( + self, playbook, spec: TaskSpec, run_dir: Path, *, llm, + resolution_mode: str = "resume", tier: str = "?", + resuming: bool = False, held_lock: RunLock | None = None, + planned_context: RepositoryContext | None = None, + ) -> ExecutionResult: + """Run a resolved `playbook` to completion in `run_dir`: init tracing + + notifier, seed the shared state (repo path, push policy, protected + branches / high-risk modules from the adapter when present), drive the + Executor, then print per-step marks, optional run metrics, and the final + status. `resuming` seeds the state so steps can pick up where they left + off. Returns the exit code and blocked reason to the application, which owns + terminal `run_status.json` writes for reserved runs.""" + blocked_reason = "" + try: + repo_context = self.repository_context.for_spec(spec) + except AdapterError as exc: + blocked_reason = f"repository context invalid: {exc}" + print(style("✋ ", "red", "bold") + blocked_reason) + return ExecutionResult(BLOCKED_EXIT, blocked_reason) + if planned_context is not None and repo_context != planned_context: + blocked_reason = "repository context changed after planning" + print(style("✋ ", "red", "bold") + blocked_reason) + return ExecutionResult(BLOCKED_EXIT, blocked_reason) + lock = held_lock + if lock is None: + try: + # First, before any trace/status write: a losing concurrent + # invocation (second --resume) must leave the active run's + # artifacts completely untouched. execute_reserved holds the + # lock across its whole lifecycle and passes it in instead. + lock = RunLock(run_dir).acquire() + except RunLockHeld as exc: + print(style("✋ ", "red", "bold") + str(exc)) + return ExecutionResult(BLOCKED_EXIT, blocked_reason) + # Per-repo knowledge lock, SHARED, held for the run's lifetime: + # never contends with other runs; it exists so the knowledge + # migration's EXCLUSIVE acquire can prove no potential store + # writer is alive (and so no run starts mid-migration). + knowledge_lock = None + if spec.repo: + from ..memory.paths import (KnowledgeLockHeld, + KnowledgePaths, + KnowledgeRunLock, + KnowledgeStateError) + try: + knowledge_lock = KnowledgeRunLock( + KnowledgePaths.resolve(self.settings, spec.repo) + .knowledge_run_lock).acquire_shared() + except (KnowledgeLockHeld, KnowledgeStateError) as exc: + # BOTH refusals take the terminal protocol: an invalid + # activation marker (KnowledgeStateError) must exit + # blocked/3 with the run lock RELEASED, never escape as + # a traceback that leaves the lock held in a long-lived + # process (PR-boundary F4) + if held_lock is None: + lock.release() + print(style("✋ ", "red", "bold") + str(exc)) + return ExecutionResult(BLOCKED_EXIT, blocked_reason) + try: + from .. import tracing + tracing.init(run_dir.name, run_dir / "trace.jsonl") + # Stamp the workflow into the span file itself, so a trace lifted out of + # its run directory still says which playbook and task produced it. + tracing.run_meta(playbook=f"{playbook.name}@{playbook.version}", + task_kind=spec.kind, repo=spec.repo, tier=tier, + mode=spec.mode, resolution=resolution_mode, + report_only=spec.report_only, post=spec.post, + resuming=resuming, params=spec.params) + trace = RunTrace(run_dir / "run_trace.jsonl") + notifier = Notifier(self.settings, run_dir, trace, run_dir.name) + trace.record("task", spec=spec.model_dump(), resolution=resolution_mode, + playbook=playbook.name, tier=tier) + state: dict = { + "task_spec": spec.model_dump(), + "repo_path": repo_context.repo_path, + "push_policy": PushPolicy(), # steps may replace with a derived policy + "protected_branches": list(repo_context.protected_branches), + "resuming": resuming, + } + if repo_context.high_risk_modules: + state["high_risk_modules"] = list(repo_context.high_risk_modules) + executor = Executor(self.registry, self.settings, run_dir=run_dir, + trace=trace, llm=llm, notifier=notifier) + # run_guarded finalizes inside the event loop: playbooks that + # register run finalizers (lifecycle.register_finalizer) get + # teardown on every exit path. Nothing registered == no-op. + outcome = asyncio.run( + run_guarded(executor.run(playbook, state), run_dir)) + + if outcome.status == "done": + notifier.resolve() + + if self.settings.metrics_enabled: + try: # metrics are facts about the run; never let them break it + from ..metrics import collect_run_metrics + m = collect_run_metrics(run_dir, self.settings, outcome.status) + print(format_metrics_line(m, run_dir)) + except Exception as exc: + trace.record("metrics_error", error=f"{type(exc).__name__}: {exc}") + + for step_id, r in outcome.step_results.items(): + mark = style("✓", "green") if r.ok else style("✗", "red", "bold") + print(f" {mark} {step_id}: {r.summary}") + print(f"run {run_dir.name}: {outcome.status} ({run_dir})") + if outcome.status == "blocked": + blocked_reason = str(outcome.blocked_reason or "") + print(style(" ⚠ ", "yellow", "bold") + f"{outcome.blocked_reason}\n see {run_dir / 'ESCALATION.md'}") + return ExecutionResult(BLOCKED_EXIT, blocked_reason) + return ExecutionResult(0 if outcome.status == "done" else 1, blocked_reason) + finally: + if knowledge_lock is not None: + knowledge_lock.release() + if held_lock is None: + lock.release() diff --git a/test/test_repository_context.py b/test/test_repository_context.py index e0a1cc7c..03eb1d61 100644 --- a/test/test_repository_context.py +++ b/test/test_repository_context.py @@ -1,9 +1,13 @@ from pathlib import Path +from types import SimpleNamespace +import subprocess +import sys import pytest from infermatrix_copilot.app.core import Copilot from infermatrix_copilot.app.repository_context import RepositoryContextResolver +from infermatrix_copilot.engine.lifecycle import RunLock from infermatrix_copilot.notify import BLOCKED_EXIT from infermatrix_copilot.task_spec import TaskSpec @@ -94,3 +98,53 @@ def test_unresolved_adapter_path_does_not_claim_checkout_capability( assert context.repo_path == "" assert "repo.path" not in context.capabilities + + +def test_workflow_execution_seeds_context_and_returns_blocked_reason( + settings, tmp_path, monkeypatch, +): + _adapter(settings, tmp_path / "ambient", branches="[main, release]") + frozen = tmp_path / "frozen" + spec = TaskSpec(kind="repo_rebase", repo_path=str(frozen)) + seen = {} + + class FakeExecutor: + def __init__(self, *args, **kwargs): + pass + + async def run(self, playbook, state): + seen.update(state) + return SimpleNamespace( + status="blocked", blocked_reason="fixture stop", step_results={}, + ) + + monkeypatch.setattr( + "infermatrix_copilot.app.workflow_execution.Executor", FakeExecutor, + ) + copilot = Copilot(settings) + run_dir = tmp_path / "run-context" + + code = copilot._execute( + SimpleNamespace(name="fixture", version=1), spec, run_dir, + planned_context=copilot.repository_context.for_spec(spec), + ) + + assert code == BLOCKED_EXIT + assert copilot.last_blocked_reason == "fixture stop" + assert seen["repo_path"] == str(frozen) + assert seen["protected_branches"] == ["main", "release"] + assert seen["high_risk_modules"] == ["scheduler"] + with RunLock(run_dir): + pass + + +def test_workflow_execution_import_does_not_load_transports(): + code = ( + "import sys\n" + "import infermatrix_copilot.app.workflow_execution\n" + "assert not any(name.startswith(('infermatrix_copilot.cli', " + "'infermatrix_copilot.mcp', 'infermatrix_copilot.thin_mcp')) " + "for name in sys.modules)\n" + ) + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) + assert result.returncode == 0, result.stderr