Skip to content

Separate workflow execution from Copilot orchestration #198

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tzhouam merged 1 commit into from
Sep 26, 2026
Merged

Separate workflow execution from Copilot orchestration #198

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions doc/architecture/SPEC/app/core.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
21 changes: 21 additions & 0 deletions doc/architecture/SPEC/app/workflow_execution.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
# app/workflow_execution.py — workflow execution boundary

<!-- verified-against: 2026-09-26 -->

`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.
135 changes: 16 additions & 119 deletions src/infermatrix_copilot/app/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
154 changes: 154 additions & 0 deletions src/infermatrix_copilot/app/workflow_execution.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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()
Loading
Loading