artifex/settings.py

@@ -24,7 +24,6 @@ INSTALLED_APPS = [
     "control_plane.secrets",
     "control_plane.knowledge",
     "control_plane.verification",
-    "graph",
 ]
 
 MIDDLEWARE = [

control_plane/agents/migrations/0005_progenysignal_graph_lineage.py

@@ -1,29 +0,0 @@
-from __future__ import annotations
-
-import django.db.models.deletion
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("agents", "0004_progenysignal"),
-        ("graph", "0003_graphapproval"),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name="progenysignal",
-            name="execution_graph_version",
-            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to="graph.executiongraphversion"),
-        ),
-        migrations.AddField(
-            model_name="progenysignal",
-            name="graph_node_run",
-            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to="graph.graphnoderun"),
-        ),
-        migrations.AddField(
-            model_name="progenysignal",
-            name="graph_run",
-            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to="graph.graphrun"),
-        ),
-    ]

control_plane/agents/models.py

@@ -88,9 +88,6 @@ class ProgenySignal(TimestampedModel):
     task = models.ForeignKey("projects.Task", on_delete=models.SET_NULL, null=True, blank=True)
     milestone = models.ForeignKey("projects.Milestone", on_delete=models.SET_NULL, null=True, blank=True)
     agent_version = models.ForeignKey(AgentVersion, on_delete=models.SET_NULL, null=True, blank=True)
-    graph_run = models.ForeignKey("graph.GraphRun", on_delete=models.SET_NULL, null=True, blank=True)
-    graph_node_run = models.ForeignKey("graph.GraphNodeRun", on_delete=models.SET_NULL, null=True, blank=True)
-    execution_graph_version = models.ForeignKey("graph.ExecutionGraphVersion", on_delete=models.SET_NULL, null=True, blank=True)
     source = models.CharField(max_length=80)
     severity = models.CharField(max_length=32, default="INFO")
     failure_category = models.CharField(max_length=80, blank=True)

control_plane/projects/migrations/0003_commitrecord_graph_run.py

@@ -1,19 +0,0 @@
-from __future__ import annotations
-
-import django.db.models.deletion
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("graph", "0002_graphnoderun_visit_index"),
-        ("projects", "0002_commitrecord_coder_commitrecord_judge_and_more"),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name="commitrecord",
-            name="graph_run",
-            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="commits", to="graph.graphrun"),
-        ),
-    ]

control_plane/projects/models.py

@@ -173,9 +173,6 @@ class CommitRecord(TimestampedModel):
     verification = models.ForeignKey(
         "verification.Verification", on_delete=models.SET_NULL, null=True, blank=True, related_name="commits"
     )
-    graph_run = models.ForeignKey(
-        "graph.GraphRun", on_delete=models.SET_NULL, null=True, blank=True, related_name="commits"
-    )
     sha = models.CharField(max_length=64)
     branch_name = models.CharField(max_length=255)
     message = models.TextField()

docs/execution_graph_architecture.md

@@ -1,204 +0,0 @@
-# Execution Graph Architecture
-
-Date: 2026-08-15
-
-## Implemented Architecture
-
-Artifex now has two intentionally separate graph concepts.
-
-Project DAG remains the canonical representation of what work exists:
-
-- `Project`
-- `Milestone`
-- `Feature`
-- `Task`
-- `TaskDependency`
-
-Execution Graph is now the canonical representation of how autonomous work is performed. It is runtime-neutral and persisted in Django/Postgres.
-
-## Graph Domain Models
-
-Implemented graph domain models:
-
-- `ExecutionGraphDefinition`
-- `ExecutionGraphVersion`
-- `GraphRun`
-- `GraphNodeRun`
-- `GraphEdgeTraversal`
-- `GraphApproval`
-
-Execution graph version statuses:
-
-- `DRAFT`
-- `CHALLENGER`
-- `CHAMPION`
-- `RETIRED`
-
-Graph run statuses:
-
-- `PENDING`
-- `RUNNING`
-- `PAUSED`
-- `COMPLETE`
-- `FAILED`
-- `CANCELLED`
-
-Node runs persist status, timing, input/output metadata, failure evidence, telemetry, optional agent version, and optional model request reference.
-
-Edge traversals persist selected conditional transitions.
-
-## Graph Specification
-
-Graph specs are serializable dictionaries with:
-
-- name
-- version
-- graph type
-- entry node
-- nodes
-- edges
-- conditional edge labels
-- terminal nodes
-- metadata
-
-Specs do not execute arbitrary Python. Node execution resolves through `NodeHandlerRegistry`.
-
-## TaskExecutionGraph V1
-
-`TaskExecutionGraph v1` mirrors the prior `AutonomousTaskLoop` behavior.
-
-Nodes:
-
-- `claim_task`
-- `prepare_worktree`
-- `build_context`
-- `coder`
-- `run_tests`
-- `review`
-- `judge`
-- `commit`
-- `retry_or_fail`
-- `cleanup`
-- `complete`
-- `fail`
-
-Important semantics preserved:
-
-- deterministic test failure still reaches Reviewer
-- Reviewer failure goes to retry/fail
-- Judge failure goes to retry/fail
-- maximum semantic task attempts remains three total attempts
-- commit is guarded against duplicate commits on resume
-- CoderToolLoop remains inside the `coder` node
-
-## NativeGraphRuntime
-
-`NativeGraphRuntime` is the reference implementation.
-
-It supports:
-
-- conditional edges
-- loops
-- persisted graph runs
-- persisted node runs
-- edge traversal history
-- checkpoint/resume at major node boundaries
-- terminal success/failure
-- pause state
-- cancellation
-- graph events
-- bounded metadata
-
-Django/Postgres remains canonical for task, attempt, worktree, test, review, judge, commit, event, agent, and Progeny state.
-
-## AutonomousTaskLoop Delegation
-
-`AutonomousTaskLoop` still uses `TaskScheduler` to claim Project DAG work.
-
-After a task is claimed, it now creates a `GraphRun` for the champion `TaskExecutionGraph v1` and delegates task lifecycle execution to `NativeGraphRuntime`.
-
-The scheduler remains Project-DAG-oriented. The execution runtime handles task-attempt workflow.
-
-## LangGraphRuntime
-
-`LangGraphRuntime` is no longer a pure placeholder. It builds a LangGraph `StateGraph` from Artifex graph definitions when the `langgraph` package is installed.
-
-Artifex remains runtime-neutral:
-
-- Artifex owns graph definitions
-- Artifex owns persisted graph state
-- Artifex owns node contracts
-- LangGraph is an execution backend
-
-Current limitation: the local environment used during implementation did not have `langgraph` installed, so deterministic tests validate the adapter boundary and missing-dependency behavior. Full LangGraph execution parity requires installing the declared `langgraph>=0.2,<0.3` dependency in local/Spark environments.
-
-## Checkpoint And Resume
-
-Coarse checkpointing is implemented at graph node boundaries.
-
-Resume behavior avoids repeating a completed interrupted node. Loop re-entry creates a new `GraphNodeRun` visit via `visit_index`.
-
-Irreversible commit behavior is guarded by checking for an existing `CommitRecord` for the task before creating a new commit.
-
-## Human Approval
-
-Graph approval support is implemented with `GraphApproval`.
-
-A node can pause the graph with `AWAITING_APPROVAL`. A signal can approve pending graph approvals and resume execution.
-
-This is intentionally minimal and prepares future gates for risky migrations, deployment, Progeny promotion, destructive infrastructure changes, and project plan approvals.
-
-## Subgraph Support
-
-The graph spec supports runtime-neutral subgraph representation through node metadata.
-
-CoderToolLoop is not moved into a subgraph yet. It remains inside the `coder` node.
-
-## Graph Inspection
-
-`graph_run_inspection()` exposes UI-ready JSON containing:
-
-- graph/version
-- node list
-- edges
-- node statuses
-- current node
-- durations
-- failures
-- selected edge traversals
-
-This is sufficient for a future visual execution graph UI.
-
-## Progeny Integration
-
-`ProgenySignal` can now reference:
-
-- `graph_run`
-- `graph_node_run`
-- `execution_graph_version`
-
-This enables future investigations such as:
-
-- failures by graph version
-- failures by node type
-- patch success rate by workflow version
-- Reviewer rework changes after inserting a verification node
-
-## Migrations
-
-New migrations:
-
-- `graph/0001_initial.py`
-- `graph/0002_graphnoderun_visit_index.py`
-- `graph/0003_graphapproval.py`
-- `projects/0003_commitrecord_graph_run.py`
-- `agents/0005_progenysignal_graph_lineage.py`
-
-## Known Limitations
-
-- Full LangGraph parity execution is implemented but not exercised in this environment because `langgraph` is not installed.
-- Parallel branch execution is represented by the graph model but not executed concurrently by `NativeGraphRuntime`.
-- Subgraph support is represented in the spec but not yet expanded into nested `GraphRun` execution.
-- CoderToolLoop remains attempt-granularity for checkpointing.
-- Graph telemetry is persisted at node level, but model request references are not yet automatically linked to individual node runs.
-- Event Bus records graph facts but is not an orchestration engine.

docs/graph_runtime_audit.md

@@ -1,524 +0,0 @@
-# Graph Runtime Audit
-
-Date: 2026-08-15
-
-Baseline inspected: `6a73372665715950aee8416a9d2b3ec79955f816`
-
-Scope: audit current Artifex implementation against the intended graph-runtime architecture. This document does not propose or include a refactor.
-
-## Executive Summary
-
-`GraphRuntime` is currently an interface boundary, not the production execution engine.
-
-`LangGraphRuntime` is a deterministic placeholder adapter. It is not executing project, milestone, feature, task, Coder, Reviewer, Judge, retry, or commit workflows.
-
-Real production task execution is driven by `runtime_loop.autonomous_task_loop.AutonomousTaskLoop`, `graph.scheduler.TaskScheduler`, Django ORM state, direct Python service calls, and persisted Event Bus events.
-
-The current implementation has two separate concepts that are easy to conflate:
-
-- Project DAG: persisted project work decomposition and task dependencies.
-- Execution Graph: the operational workflow for executing one task attempt through Coder, tools, tests, Reviewer, Judge, retry, and commit.
-
-The Project DAG exists today in Django models and scheduler logic. The Execution Graph exists only as imperative Python control flow inside `AutonomousTaskLoop`; it is not represented as a graph runtime yet.
-
-Recommendation: keep Django/Postgres as canonical state and migrate execution to graph semantics incrementally. Start by extracting an explicit `TaskExecutionGraph` description around the existing task loop without changing behavior, then add graph telemetry/checkpoint metadata. LangGraph should remain behind `GraphRuntime` and should provide value only when conditional routing, resumable execution, human gates, subgraphs, and agent-team workflows become difficult to manage in plain Python.
-
-## Direct Answers
-
-### 1. Is `GraphRuntime` used in production task execution?
-
-No.
-
-`graph/runtime.py` defines an abstract `GraphRuntime` with `start`, `pause`, `resume`, `cancel`, and `signal`, but no production task execution path depends on it.
-
-Observed production path:
-
-- `runtime_loop/autonomous_task_loop.py` imports `TaskScheduler`, `Coder`, `Reviewer`, `Judge`, `WorktreeTools`, and `DeterministicTestRunner`.
-- It does not import or call `GraphRuntime` or `LangGraphRuntime`.
-
-Current role of `GraphRuntime`: architectural boundary and future abstraction.
-
-### 2. Is `LangGraphRuntime` actually executing task workflows?
-
-No.
-
-`graph/langgraph_runtime.py` returns a deterministic string from `start` and no-ops `pause`, `resume`, `cancel`, and `signal`.
-
-It does not import LangGraph, build a state graph, execute graph nodes, route edges, persist graph checkpoints, or call Coder/Reviewer/Judge.
-
-Its only observed test coverage is `tests/test_interfaces.py::test_langgraph_runtime_is_hidden_behind_runtime_interface`, which checks that `start()` returns a run id prefix.
-
-## Real Execution Path
-
-Current task execution path for a single task:
-
-1. Scheduler
-
-`AutonomousTaskLoop.run_once()` calls `TaskScheduler.claim_next_ready_task()`.
-
-`TaskScheduler` queries `Task(status=READY)` ordered by priority and creation time, skips tasks with incomplete dependencies, marks the selected task `RUNNING`, persists it, and publishes `TASK_STARTED`.
-
-2. Investigation/context building
-
-`AutonomousTaskLoop._execute_task()` creates or reuses a worktree and builds context with `WorkerContextBuilder.build_for_task()`.
-
-This context builder reads selected Python files and tests directly from the worktree, excludes migrations and secrets, and attaches the current diff.
-
-Coder also has an inspect-before-edit phase inside `Coder.execute()` for sensitive tasks. It asks the model for read-only inspection operations, then executes `list_directory`, `read_file`, `search_code`, `find_symbol`, `git_status`, or `git_diff` through `WorktreeTools`.
-
-3. Coder
-
-`AutonomousTaskLoop` calls `Coder.execute(context, tools, project, agent_version)`.
-
-`Coder.execute()` may perform inspection, then creates a `CoderToolLoop` and calls `CoderToolLoop.run()`.
-
-4. Mutation
-
-`CoderToolLoop` asks the model for mutation operations and applies them through `WorktreeTools`.
-
-Supported mutation operations include:
-
-- `write_file`
-- `apply_patch`
-- `delete_file`
-- `move_file`
-- `create_directory`
-- `run_command`
-
-Patch mismatch recovery is internal to `CoderToolLoop`: it records structured failure evidence, re-reads the live target file excerpt, and asks the model once more for a corrected patch. It does not default to full-file replacement.
-
-5. Tests
-
-`AutonomousTaskLoop` calls `DeterministicTestRunner.run()`.
-
-`DeterministicTestRunner` shells out via `WorktreeTools.run()`, persists an `Artifact` with stdout/stderr/returncode, and creates a `TestRun` with `PASS` or `FAIL`.
-
-On failure, `_attach_test_failure_evidence()` stores stdout/stderr excerpts into the attempt's `coder_result.metadata.test_failure_evidence`, so later retry context can include the actual traceback.
-
-6. Reviewer
-
-`AutonomousTaskLoop` stages intent-to-add with `git add -N`, gets `git diff`, then calls `Reviewer.review(task, reviewer_version, diff, test_status)`.
-
-`Reviewer` is currently a deterministic Python policy. It creates a `Review` row with `PASS`, `REWORK_REQUIRED`, or `REJECTED` based on test status, empty diff, and health-route-specific heuristics.
-
-7. Judge
-
-If Reviewer passes, `AutonomousTaskLoop` calls `Judge.judge(project, task, judge_version, diff, test_status)`.
-
-`Judge` is currently deterministic Python. It creates a `Verification` row with task-level acceptance evidence and `PASS` or `FAIL`.
-
-8. Retry
-
-If Coder, Reviewer, or Judge fails, `_retry_or_fail()` updates `TaskAttempt.status`, increments `Task.retry_count`, publishes `TASK_FAILED`, and either loops for another attempt or marks the task `FAILED` and publishes `TASK_RETRY_EXHAUSTED` plus final `TASK_FAILED`.
-
-The effective retry budget is capped at two retries, giving three total attempts.
-
-9. Commit
-
-If tests, Reviewer, and Judge all pass, `WorktreeTools.commit_all()` stages and commits all changes. `CommitRecord` is persisted with coder, reviewer, judge, test run, review, verification, SHA, branch, and message.
-
-The task is marked `COMPLETE`, `COMMIT_CREATED` and `TASK_COMPLETED` are published with telemetry, the worktree is validated clean, and the task worktree is cleaned up.
-
-## Transition Classification
-
-### Explicit graph nodes/edges
-
-Currently only the Project DAG has explicit persisted edges:
-
-- `TaskDependency(task, depends_on)` models task-to-task dependencies.
-- `Milestone.dependencies` models milestone-level dependencies but is not currently central in execution scheduling.
-- `ProjectPlanBuilder` validates task dependency acyclicity before materializing tasks.
-- `TaskScheduler.claim_next_ready_task()` treats incomplete `TaskDependency` rows as blocking edges.
-
-No explicit graph nodes/edges currently model the execution workflow inside a task attempt.
-
-### Direct Python service calls
-
-Most real execution transitions are direct service calls:
-
-- `AutonomousTaskLoop.run_once()` -> `TaskScheduler.claim_next_ready_task()`
-- `_execute_task()` -> `WorkerContextBuilder.build_for_task()`
-- `_execute_task()` -> `Coder.execute()`
-- `Coder.execute()` -> `CoderToolLoop.run()`
-- `CoderToolLoop` -> `ModelRouter.complete()`
-- `CoderToolLoop` -> `WorktreeTools` mutation/read/test helpers
-- `_execute_task()` -> `DeterministicTestRunner.run()`
-- `_execute_task()` -> `Reviewer.review()`
-- `_execute_task()` -> `Judge.judge()`
-- `_execute_task()` -> `WorktreeTools.commit_all()`
-- `_retry_or_fail()` direct retry/fail transition
-
-### Django state transitions
-
-Django/Postgres is the canonical durable state machine today:
-
-- `Task.status`: `READY`, `RUNNING`, `FAILED`, `COMPLETE`
-- `Task.retry_count`
-- `TaskAttempt.status`: `RUNNING`, `REWORK_REQUIRED`, `FAILED`, `COMPLETE`
-- `Worktree.status`
-- `TestRun.status`
-- `Review.status`
-- `Verification.result`
-- `CommitRecord`
-- `ModelRequest.status`
-- `ProgenySignal.status`
-
-### Event Bus driven transitions
-
-The Event Bus persists facts but does not currently drive the control flow.
-
-Events published today include:
-
-- `PLAN_APPROVED`
-- `MILESTONE_CREATED`
-- `TASK_CREATED`
-- `TASK_READY`
-- `TASK_STARTED`
-- `TEST_FAILED`
-- `REVIEW_FAILED`
-- `TASK_FAILED`
-- `TASK_RETRY_EXHAUSTED` as a string event
-- `COMMIT_CREATED`
-- `TASK_COMPLETED`
-- `AGENT_CREATED`
-- `AGENT_PROMOTED`
-- `PROGENY_SIGNAL_CREATED` as a string event from `ProgenyService`
-
-These are audit/event facts, not asynchronous orchestration triggers in the current implementation.
-
-## Project DAG vs Execution Graph
-
-### Project DAG
-
-The Project DAG answers: what work exists, and what depends on what?
-
-Current implementation:
-
-- `ProjectPlanContract` represents milestones, features, tasks, and task dependency references.
-- `validate_project_plan()` validates task dependency references and cycle freedom.
-- `ProjectPlanBuilder.apply()` persists `ProjectPlan`, `Milestone`, `Feature`, `Task`, and `TaskDependency` rows.
-- `TaskScheduler` consumes `TaskDependency` rows to claim only unblocked `READY` tasks.
-
-This is real and production-relevant today.
-
-### Execution Graph
-
-The Execution Graph answers: how is one task/feature/milestone autonomously executed?
-
-Current implementation:
-
-- This graph is implicit in `AutonomousTaskLoop._execute_task()`.
-- Nodes are not persisted as graph nodes.
-- Edges are Python `if`/`continue`/`return` branches.
-- Looping is a Python `while task.retry_count <= effective_max_retries`.
-- Checkpointing is coarse-grained through Django rows such as `TaskAttempt`, `TestRun`, `Review`, `Verification`, and `CommitRecord`.
-
-The execution graph exists conceptually but not as a first-class runtime artifact.
-
-## CoderToolLoop Placement
-
-Recommended placement: graph node containing an internal tool loop.
-
-Do not make `CoderToolLoop` the top-level execution graph yet.
-
-Rationale:
-
-- `CoderToolLoop` is a bounded inner loop for a single agent/tool interaction.
-- Its state is local and tactical: inspection results, mutation results, patch mismatch recovery, operation telemetry.
-- It should be resumable eventually, but it is not the same graph as the whole task lifecycle.
-- Making every tool call a top-level graph node now would increase orchestration complexity before there is enough operational pressure.
-
-Near-term target:
-
-- `TaskExecutionGraph` has a `coder` node.
-- The `coder` node invokes `Coder.execute()`.
-- `Coder.execute()` owns `CoderToolLoop` internally.
-- `CoderToolLoop` emits structured telemetry and failure evidence that the outer graph records.
-
-Later target:
-
-- If long-running coding sessions, human approvals, streamed model/tool execution, or partial resume become necessary, promote internal tool-loop steps to a nested subgraph.
-
-## Runtime Capability Assessment
-
-### Conditional edges
-
-Required.
-
-Current conditional branches include:
-
-- no ready task -> stop
-- dependencies incomplete -> skip task
-- coder failed -> retry/fail
-- tests failed -> still Review, then retry/fail through Review
-- reviewer failed -> retry/fail
-- judge failed -> retry/fail
-- all checks passed -> commit
-
-These map naturally to graph conditional edges.
-
-### Loops
-
-Required.
-
-Current loops include:
-
-- task retry loop capped at three total attempts
-- CoderToolLoop patch recovery loop
-- future investigation and experiment loops
-
-The retry cap must remain durable and enforced by Django state, not hidden only inside graph runtime memory.
-
-### Parallel branches
-
-Useful, not urgent for current V1/V2 guarantees.
-
-Good candidates later:
-
-- independent task execution across the Project DAG
-- parallel tests/static checks/security checks
-- multi-agent review lanes
-- Progeny benchmark replay across candidates
-
-Parallelism should be introduced only after worktree isolation and resource contention controls are explicit.
-
-### Subgraphs
-
-Required long-term.
-
-Natural subgraphs:
-
-- Task execution
-- Coder tool loop
-- Reviewer/Judge verification
-- Progeny investigation
-- Progeny experiment/champion-challenger evaluation
-- Project Brain planning/replanning
-
-### Checkpoint/resume
-
-Required, but Django/Postgres should remain canonical.
-
-Current checkpointing already exists at coarse boundaries:
-
-- task claimed
-- attempt created
-- coder result saved
-- test run saved
-- review saved
-- verification saved
-- commit saved
-
-Graph checkpointing should augment these records, not replace them.
-
-### Human approval nodes
-
-Required for safe V2+ autonomy.
-
-Likely approval gates:
-
-- project plan approval
-- task split/replan approval
-- risky migration approval
-- provider credential/resource approval
-- promotion of Progeny candidates
-- destructive infrastructure changes
-
-Human approval should persist as Django state and be represented as paused graph nodes.
-
-### Dynamic temporary agent teams
-
-Required later, not required for the current loop.
-
-Smart Investigation and Progeny experiments will likely need temporary teams such as:
-
-- incident analyst
-- domain expert
-- prompt/policy editor
-- benchmark designer
-- reviewer challenger
-
-The runtime should support this as graph metadata and persisted `AgentRun`/future team records.
-
-### Graph versioning
-
-Required before graph-based production execution.
-
-Every accepted commit and Progeny investigation should know which execution graph version produced it. Without this, regressions in orchestration will be hard to diagnose.
-
-### Graph telemetry
-
-Required.
-
-Existing telemetry should become graph-node telemetry:
-
-- mutation operations
-- patch success/mismatch rates
-- write-file fallback rate
-- model requests per task
-- test duration/status
-- review findings
-- judge evidence
-- retry count
-- provider errors
-- signal counts
-
-This telemetry should feed Progeny.
-
-### Progeny champion/challenger workflow graphs
-
-Required for the future Replay Arena / experiment execution layer, but should not be the next refactor.
-
-The Progeny workflow graph should model:
-
-- investigation selection
-- candidate generation
-- benchmark construction
-- champion/challenger replay
-- judge/evaluator comparison
-- approval/promotion/rejection
-
-It should reuse the same runtime boundary but have separate graph versions from normal task execution.
-
-## Where LangGraph Adds Material Value
-
-LangGraph is likely valuable when Artifex needs durable, inspectable, conditional, resumable multi-agent workflows with nested loops.
-
-Material-value areas:
-
-- Explicit task execution graph with conditional retry/fail/commit branches.
-- Smart Investigation graph: signals -> clusters -> hypotheses -> routing -> candidates.
-- Human approval gates with pause/resume semantics.
-- Progeny champion/challenger workflows.
-- Dynamic temporary agent-team orchestration.
-- Future parallel branches for tests/reviews/replays.
-- Graph-level observability and visual traceability.
-
-Low-value or risky areas for LangGraph right now:
-
-- Replacing `TaskScheduler` as the canonical Project DAG scheduler.
-- Replacing Django/Postgres state transitions.
-- Turning every small Python helper call into a graph node.
-- Moving worktree mutation safety into graph runtime abstractions.
-- Hiding retry counters/checkpoints inside LangGraph-only state.
-
-## Where Django/Postgres Should Remain Canonical
-
-Django/Postgres should remain the source of truth for:
-
-- Projects, plans, milestones, features, tasks, and task dependencies.
-- Task statuses and retry counts.
-- Worktrees and commit records.
-- Attempts, coder results, reviews, judge verifications, test runs, artifacts.
-- Events.
-- Agent versions and promotion state.
-- Progeny signals, investigations, and improvement candidates.
-- Human approvals and audit history.
-
-LangGraph, if introduced, should read/write through these canonical records rather than owning hidden state.
-
-## Migration Path
-
-### Phase 0: Preserve current behavior
-
-Do not route production task execution through `LangGraphRuntime` yet.
-
-Current guarantees to preserve:
-
-- M2 autonomous execution
-- M3 planning
-- M4 Archaeologist
-- M5 Agent Registry / Progeny
-- M6 self-bootstrap
-- CoderToolLoop
-- worktree isolation
-- inspect-before-edit
-- safe mutation tools
-- Reviewer/Judge separation
-- deterministic tests
-- three-attempt retry cap
-- Sol/Qwen provider boundaries
-- secret isolation
-
-### Phase 1: Document the implicit execution graph
-
-Add a declarative graph spec that mirrors current `AutonomousTaskLoop` behavior without executing it yet.
-
-Suggested nodes:
-
-- `claim_task`
-- `prepare_worktree`
-- `build_context`
-- `coder`
-- `run_tests`
-- `review`
-- `judge`
-- `commit`
-- `retry_or_fail`
-- `cleanup`
-
-Suggested conditional edges:
-
-- `coder.failed -> retry_or_fail`
-- `tests.failed -> review`
-- `review.failed -> retry_or_fail`
-- `judge.failed -> retry_or_fail`
-- `judge.passed -> commit`
-- `retry.available -> build_context`
-- `retry.exhausted -> fail`
-
-### Phase 2: Add graph run telemetry around the existing loop
-
-Add persisted graph-run/node-run records or event payloads while keeping direct Python execution.
-
-This gives observability before changing orchestration.
-
-### Phase 3: Introduce `TaskExecutionRuntime` behind `GraphRuntime`
-
-Implement a runtime adapter that can execute the graph spec using the existing service methods.
-
-At this stage, `AutonomousTaskLoop` can delegate to the runtime but Django state remains canonical.
-
-### Phase 4: Evaluate LangGraph as one adapter
-
-Only after the explicit graph spec and telemetry exist, wire `LangGraphRuntime` as an implementation option.
-
-Acceptance criteria should require deterministic parity with the current loop.
-
-### Phase 5: Add subgraphs and human approval nodes
-
-Promote Progeny investigation and champion/challenger replay workflows into graph subgraphs.
-
-Add explicit approval nodes for candidate promotion and risky changes.
-
-## Current Architecture Risks
-
-1. `GraphRuntime` naming implies production graph execution, but it is currently a placeholder.
-
-Risk: future contributors may assume LangGraph is already in the hot path.
-
-2. Execution graph is implicit in Python.
-
-Risk: retry and failure behavior becomes harder to evolve as Progeny wiring, investigation, and human gates are added.
-
-3. Event Bus is persisted but not orchestration-driving.
-
-Risk: event names may be interpreted as workflow triggers when they are currently audit facts.
-
-4. `CoderToolLoop` has internal iteration but only coarse persistence.
-
-Risk: power loss during model/tool-loop execution resumes only at task-attempt granularity, not tool-step granularity.
-
-5. Reviewer/Judge are deterministic services, not model-driven graph nodes.
-
-Risk: current separation is good, but future multi-agent review designs need explicit graph boundaries.
-
-## Recommendation
-
-Do not refactor execution into LangGraph immediately.
-
-The safest next step is to keep the existing Python/Django loop as the production engine and create a first-class, documented `TaskExecutionGraph` specification plus telemetry records around it. This preserves all working guarantees while making the intended graph architecture concrete.
-
-CoderToolLoop should remain an internal loop inside the Coder graph node for now. Promote it to a subgraph only when resumability or multi-step observability at each tool call becomes necessary.
-
-LangGraph should be introduced as an adapter behind `GraphRuntime` only after deterministic parity tests prove it reproduces the current loop exactly.

graph/apps.py

@@ -1,8 +0,0 @@
-from __future__ import annotations
-
-from django.apps import AppConfig
-
-
-class GraphConfig(AppConfig):
-    default_auto_field = "django.db.models.BigAutoField"
-    name = "graph"

graph/bootstrap.py

@@ -1,29 +0,0 @@
-from __future__ import annotations
-
-from django.utils import timezone
-
-from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus
-from graph.task_execution import task_execution_graph_v1
-
-
-def champion_task_execution_graph_v1() -> ExecutionGraphVersion:
-    spec = task_execution_graph_v1()
-    definition, _ = ExecutionGraphDefinition.objects.get_or_create(
-        name=spec.name,
-        defaults={"graph_type": spec.graph_type, "description": str(spec.metadata.get("description", ""))},
-    )
-    version, created = ExecutionGraphVersion.objects.get_or_create(
-        graph=definition,
-        version=spec.version,
-        defaults={
-            "status": ExecutionGraphVersionStatus.CHAMPION,
-            "graph_spec": spec.to_dict(),
-            "metadata": {"immutable_after_use": True},
-            "promoted_at": timezone.now(),
-        },
-    )
-    if not created and version.status != ExecutionGraphVersionStatus.CHAMPION:
-        version.status = ExecutionGraphVersionStatus.CHAMPION
-        version.promoted_at = timezone.now()
-        version.save(update_fields=["status", "promoted_at"])
-    return version

graph/inspection.py

@@ -1,37 +0,0 @@
-from __future__ import annotations
-
-from graph.models import GraphRun
-
-
-def graph_run_inspection(graph_run: GraphRun) -> dict[str, object]:
-    spec = graph_run.execution_graph_version.graph_spec
-    node_runs = {
-        node.node_id: node
-        for node in graph_run.node_runs.order_by("node_id", "-visit_index")
-    }
-    nodes = []
-    for node_id, node_spec in spec.get("nodes", {}).items():
-        run = node_runs.get(node_id)
-        nodes.append(
-            {
-                "id": node_id,
-                "type": node_spec.get("type"),
-                "status": run.status if run else "PENDING",
-                "visit_index": run.visit_index if run else 0,
-                "duration_ms": (run.telemetry or {}).get("duration_ms") if run else None,
-                "failure": run.failure_evidence if run else {},
-                "metadata": node_spec.get("metadata", {}),
-            }
-        )
-    return {
-        "graph": graph_run.execution_graph_version.graph.name,
-        "version": graph_run.execution_graph_version.version,
-        "status": graph_run.status,
-        "current_node": graph_run.current_node,
-        "nodes": nodes,
-        "edges": spec.get("edges", []),
-        "edge_traversals": list(
-            graph_run.edge_traversals.order_by("created_at").values("source_node", "target_node", "condition", "result", "metadata", "created_at")
-        ),
-        "failures": list(graph_run.node_runs.exclude(failure_evidence={}).values("node_id", "failure_evidence")),
-    }

graph/langgraph_runtime.py

@@ -1,95 +1,29 @@
 from __future__ import annotations
 
-from typing import Any, TypedDict
+from typing import Any
 from uuid import UUID
 
-from graph.models import GraphRun
-from graph.native_runtime import NativeGraphRuntime
-from graph.registry import NodeHandlerRegistry
 from graph.runtime import GraphRuntime
-from graph.spec import ExecutionGraphSpec
-
-
-class LangGraphState(TypedDict, total=False):
-    graph_run_id: int
-    current_node: str
-    status: str
-    edge_result: str
 
 
 class LangGraphRuntime(GraphRuntime):
-    """LangGraph adapter behind Artifex's runtime-neutral graph boundary."""
+    """Initial LangGraph adapter placeholder.
 
-    def __init__(self, registry: NodeHandlerRegistry | None = None) -> None:
-        self.registry = registry
+    M1 keeps this deterministic. M2 will wire the autonomous task loop here while
+    preserving this boundary.
+    """
 
-    async def start(self, project_id: UUID | None = None, **kwargs: Any) -> str:
-        graph_run = kwargs.get("graph_run")
-        if graph_run is not None:
-            self.run_until_terminal_or_paused(graph_run)
-            return str(graph_run.id)
+    async def start(self, project_id: UUID) -> str:
         return f"project-{project_id}"
 
     async def pause(self, run_id: str) -> None:
         return None
 
     async def resume(self, run_id: str) -> None:
-        if self.registry is None:
         return None
-        graph_run = GraphRun.objects.get(id=run_id)
-        self.run_until_terminal_or_paused(graph_run)
 
     async def cancel(self, run_id: str) -> None:
         return None
 
     async def signal(self, run_id: str, event: dict[str, Any]) -> None:
         return None
-
-    def run_until_terminal_or_paused(self, graph_run: GraphRun) -> GraphRun:
-        if self.registry is None:
-            raise RuntimeError("LangGraphRuntime requires a NodeHandlerRegistry")
-        try:
-            from langgraph.graph import END, StateGraph
-        except ImportError as exc:
-            raise RuntimeError("LangGraphRuntime requires the langgraph package") from exc
-
-        spec = ExecutionGraphSpec.from_dict(graph_run.execution_graph_version.graph_spec)
-        native = NativeGraphRuntime(self.registry)
-
-        workflow = StateGraph(LangGraphState)
-        for node_id in spec.nodes:
-            workflow.add_node(node_id, self._node_runner(native, graph_run, node_id))
-        workflow.set_entry_point(graph_run.current_node or spec.entry)
-        for node_id in spec.nodes:
-            if node_id in spec.terminal_nodes:
-                workflow.add_edge(node_id, END)
-                continue
-            edges = [edge for edge in spec.edges if edge.source == node_id]
-            if not edges:
-                workflow.add_edge(node_id, END)
-                continue
-            workflow.add_conditional_edges(
-                node_id,
-                lambda state: str(state.get("edge_result", "success")),
-                {edge.condition or "success": edge.target for edge in edges},
-            )
-        compiled = workflow.compile()
-        compiled.invoke({"graph_run_id": graph_run.id, "current_node": graph_run.current_node or spec.entry})
-        graph_run.refresh_from_db()
-        return graph_run
-
-    def _node_runner(self, native: NativeGraphRuntime, graph_run: GraphRun, node_id: str):
-        def run_node(state: LangGraphState) -> LangGraphState:
-            if graph_run.current_node != node_id:
-                graph_run.current_node = node_id
-                graph_run.save(update_fields=["current_node", "updated_at"])
-            native.run_until_terminal_or_paused(graph_run, interrupt_after=node_id)
-            graph_run.refresh_from_db()
-            return {
-                "graph_run_id": graph_run.id,
-                "current_node": graph_run.current_node,
-                "status": graph_run.status,
-                "edge_result": str(graph_run.metadata.get("last_edge_result", "success")),
-            }
-
-        return run_node

graph/migrations/0001_initial.py

@@ -1,100 +0,0 @@
-from __future__ import annotations
-
-import django.db.models.deletion
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    initial = True
-
-    dependencies = [
-        ("agents", "0004_progenysignal"),
-        ("projects", "0002_commitrecord_coder_commitrecord_judge_and_more"),
-        ("resources", "0002_modelrequest_completion_tokens_and_more"),
-    ]
-
-    operations = [
-        migrations.CreateModel(
-            name="ExecutionGraphDefinition",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("name", models.CharField(max_length=160, unique=True)),
-                ("graph_type", models.CharField(max_length=80)),
-                ("description", models.TextField(blank=True)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("updated_at", models.DateTimeField(auto_now=True)),
-            ],
-        ),
-        migrations.CreateModel(
-            name="ExecutionGraphVersion",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("version", models.PositiveIntegerField()),
-                ("status", models.CharField(choices=[("DRAFT", "Draft"), ("CHALLENGER", "Challenger"), ("CHAMPION", "Champion"), ("RETIRED", "Retired")], default="DRAFT", max_length=32)),
-                ("graph_spec", models.JSONField(default=dict)),
-                ("metadata", models.JSONField(blank=True, default=dict)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("promoted_at", models.DateTimeField(blank=True, null=True)),
-                ("graph", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="versions", to="graph.executiongraphdefinition")),
-            ],
-        ),
-        migrations.CreateModel(
-            name="GraphRun",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("status", models.CharField(choices=[("PENDING", "Pending"), ("RUNNING", "Running"), ("PAUSED", "Paused"), ("COMPLETE", "Complete"), ("FAILED", "Failed"), ("CANCELLED", "Cancelled")], default="PENDING", max_length=32)),
-                ("started_at", models.DateTimeField(blank=True, null=True)),
-                ("completed_at", models.DateTimeField(blank=True, null=True)),
-                ("current_node", models.CharField(blank=True, max_length=120)),
-                ("failure_reason", models.TextField(blank=True)),
-                ("metadata", models.JSONField(blank=True, default=dict)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("updated_at", models.DateTimeField(auto_now=True)),
-                ("execution_graph_version", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="runs", to="graph.executiongraphversion")),
-                ("feature", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_runs", to="projects.feature")),
-                ("milestone", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_runs", to="projects.milestone")),
-                ("project", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name="graph_runs", to="projects.project")),
-                ("task", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_runs", to="projects.task")),
-                ("task_attempt", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_runs", to="projects.taskattempt")),
-            ],
-        ),
-        migrations.CreateModel(
-            name="GraphNodeRun",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("node_id", models.CharField(max_length=120)),
-                ("node_type", models.CharField(max_length=120)),
-                ("status", models.CharField(choices=[("PENDING", "Pending"), ("RUNNING", "Running"), ("PAUSED", "Paused"), ("COMPLETE", "Complete"), ("FAILED", "Failed"), ("SKIPPED", "Skipped")], default="PENDING", max_length=32)),
-                ("started_at", models.DateTimeField(blank=True, null=True)),
-                ("completed_at", models.DateTimeField(blank=True, null=True)),
-                ("input_metadata", models.JSONField(blank=True, default=dict)),
-                ("output_metadata", models.JSONField(blank=True, default=dict)),
-                ("failure_evidence", models.JSONField(blank=True, default=dict)),
-                ("telemetry", models.JSONField(blank=True, default=dict)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("updated_at", models.DateTimeField(auto_now=True)),
-                ("agent_version", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_node_runs", to="agents.agentversion")),
-                ("graph_run", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="node_runs", to="graph.graphrun")),
-                ("model_request", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="graph_node_runs", to="resources.modelrequest")),
-            ],
-        ),
-        migrations.CreateModel(
-            name="GraphEdgeTraversal",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("source_node", models.CharField(max_length=120)),
-                ("target_node", models.CharField(max_length=120)),
-                ("condition", models.CharField(blank=True, max_length=120)),
-                ("result", models.CharField(blank=True, max_length=120)),
-                ("metadata", models.JSONField(blank=True, default=dict)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("graph_run", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="edge_traversals", to="graph.graphrun")),
-            ],
-        ),
-        migrations.AddConstraint(model_name="executiongraphversion", constraint=models.UniqueConstraint(fields=("graph", "version"), name="unique_execution_graph_version")),
-        migrations.AddConstraint(model_name="graphnoderun", constraint=models.UniqueConstraint(fields=("graph_run", "node_id"), name="unique_graph_node_run")),
-        migrations.AddIndex(model_name="graphrun", index=models.Index(fields=["status", "created_at"], name="graph_graph_status_11d04c_idx")),
-        migrations.AddIndex(model_name="graphrun", index=models.Index(fields=["task", "status"], name="graph_graph_task_id_c7ed9d_idx")),
-        migrations.AddIndex(model_name="graphnoderun", index=models.Index(fields=["node_id", "status"], name="graph_graph_node_id_43f921_idx")),
-        migrations.AddIndex(model_name="graphedgetraversal", index=models.Index(fields=["graph_run", "created_at"], name="graph_graph_graph_r_42d7dc_idx")),
-    ]

graph/migrations/0002_graphnoderun_visit_index.py

@@ -1,18 +0,0 @@
-from __future__ import annotations
-
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("graph", "0001_initial"),
-    ]
-
-    operations = [
-        migrations.RemoveConstraint(model_name="graphnoderun", name="unique_graph_node_run"),
-        migrations.AddField(model_name="graphnoderun", name="visit_index", field=models.PositiveIntegerField(default=1)),
-        migrations.AddConstraint(
-            model_name="graphnoderun",
-            constraint=models.UniqueConstraint(fields=("graph_run", "node_id", "visit_index"), name="unique_graph_node_run_visit"),
-        ),
-    ]

graph/migrations/0003_graphapproval.py

@@ -1,29 +0,0 @@
-from __future__ import annotations
-
-import django.db.models.deletion
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("graph", "0002_graphnoderun_visit_index"),
-    ]
-
-    operations = [
-        migrations.CreateModel(
-            name="GraphApproval",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("status", models.CharField(choices=[("PENDING", "Pending"), ("APPROVED", "Approved"), ("REJECTED", "Rejected")], default="PENDING", max_length=32)),
-                ("reason", models.CharField(max_length=160)),
-                ("payload", models.JSONField(blank=True, default=dict)),
-                ("requested_by", models.CharField(default="graph_runtime", max_length=120)),
-                ("decided_by", models.CharField(blank=True, max_length=120)),
-                ("decided_at", models.DateTimeField(blank=True, null=True)),
-                ("created_at", models.DateTimeField(auto_now_add=True)),
-                ("updated_at", models.DateTimeField(auto_now=True)),
-                ("graph_run", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="approvals", to="graph.graphrun")),
-                ("node_run", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name="approvals", to="graph.graphnoderun")),
-            ],
-        ),
-    ]

graph/models.py

@@ -1,129 +0,0 @@
-from __future__ import annotations
-
-from django.db import models
-
-
-class ExecutionGraphVersionStatus(models.TextChoices):
-    DRAFT = "DRAFT"
-    CHALLENGER = "CHALLENGER"
-    CHAMPION = "CHAMPION"
-    RETIRED = "RETIRED"
-
-
-class GraphRunStatus(models.TextChoices):
-    PENDING = "PENDING"
-    RUNNING = "RUNNING"
-    PAUSED = "PAUSED"
-    COMPLETE = "COMPLETE"
-    FAILED = "FAILED"
-    CANCELLED = "CANCELLED"
-
-
-class GraphNodeRunStatus(models.TextChoices):
-    PENDING = "PENDING"
-    RUNNING = "RUNNING"
-    PAUSED = "PAUSED"
-    COMPLETE = "COMPLETE"
-    FAILED = "FAILED"
-    SKIPPED = "SKIPPED"
-
-
-class ExecutionGraphDefinition(models.Model):
-    name = models.CharField(max_length=160, unique=True)
-    graph_type = models.CharField(max_length=80)
-    description = models.TextField(blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    updated_at = models.DateTimeField(auto_now=True)
-
-    def __str__(self) -> str:
-        return self.name
-
-
-class ExecutionGraphVersion(models.Model):
-    graph = models.ForeignKey(ExecutionGraphDefinition, on_delete=models.CASCADE, related_name="versions")
-    version = models.PositiveIntegerField()
-    status = models.CharField(max_length=32, choices=ExecutionGraphVersionStatus.choices, default=ExecutionGraphVersionStatus.DRAFT)
-    graph_spec = models.JSONField(default=dict)
-    metadata = models.JSONField(default=dict, blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    promoted_at = models.DateTimeField(null=True, blank=True)
-
-    class Meta:
-        constraints = [models.UniqueConstraint(fields=["graph", "version"], name="unique_execution_graph_version")]
-
-    def __str__(self) -> str:
-        return f"{self.graph.name} v{self.version}"
-
-
-class GraphRun(models.Model):
-    execution_graph_version = models.ForeignKey(ExecutionGraphVersion, on_delete=models.PROTECT, related_name="runs")
-    project = models.ForeignKey("projects.Project", on_delete=models.CASCADE, null=True, blank=True, related_name="graph_runs")
-    milestone = models.ForeignKey("projects.Milestone", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_runs")
-    feature = models.ForeignKey("projects.Feature", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_runs")
-    task = models.ForeignKey("projects.Task", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_runs")
-    task_attempt = models.ForeignKey("projects.TaskAttempt", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_runs")
-    status = models.CharField(max_length=32, choices=GraphRunStatus.choices, default=GraphRunStatus.PENDING)
-    started_at = models.DateTimeField(null=True, blank=True)
-    completed_at = models.DateTimeField(null=True, blank=True)
-    current_node = models.CharField(max_length=120, blank=True)
-    failure_reason = models.TextField(blank=True)
-    metadata = models.JSONField(default=dict, blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    updated_at = models.DateTimeField(auto_now=True)
-
-    class Meta:
-        indexes = [models.Index(fields=["status", "created_at"]), models.Index(fields=["task", "status"])]
-
-
-class GraphNodeRun(models.Model):
-    graph_run = models.ForeignKey(GraphRun, on_delete=models.CASCADE, related_name="node_runs")
-    node_id = models.CharField(max_length=120)
-    node_type = models.CharField(max_length=120)
-    visit_index = models.PositiveIntegerField(default=1)
-    status = models.CharField(max_length=32, choices=GraphNodeRunStatus.choices, default=GraphNodeRunStatus.PENDING)
-    started_at = models.DateTimeField(null=True, blank=True)
-    completed_at = models.DateTimeField(null=True, blank=True)
-    agent_version = models.ForeignKey("agents.AgentVersion", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_node_runs")
-    model_request = models.ForeignKey("resources.ModelRequest", on_delete=models.SET_NULL, null=True, blank=True, related_name="graph_node_runs")
-    input_metadata = models.JSONField(default=dict, blank=True)
-    output_metadata = models.JSONField(default=dict, blank=True)
-    failure_evidence = models.JSONField(default=dict, blank=True)
-    telemetry = models.JSONField(default=dict, blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    updated_at = models.DateTimeField(auto_now=True)
-
-    class Meta:
-        constraints = [models.UniqueConstraint(fields=["graph_run", "node_id", "visit_index"], name="unique_graph_node_run_visit")]
-        indexes = [models.Index(fields=["node_id", "status"])]
-
-
-class GraphEdgeTraversal(models.Model):
-    graph_run = models.ForeignKey(GraphRun, on_delete=models.CASCADE, related_name="edge_traversals")
-    source_node = models.CharField(max_length=120)
-    target_node = models.CharField(max_length=120)
-    condition = models.CharField(max_length=120, blank=True)
-    result = models.CharField(max_length=120, blank=True)
-    metadata = models.JSONField(default=dict, blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-
-    class Meta:
-        indexes = [models.Index(fields=["graph_run", "created_at"])]
-
-
-class GraphApprovalStatus(models.TextChoices):
-    PENDING = "PENDING"
-    APPROVED = "APPROVED"
-    REJECTED = "REJECTED"
-
-
-class GraphApproval(models.Model):
-    graph_run = models.ForeignKey(GraphRun, on_delete=models.CASCADE, related_name="approvals")
-    node_run = models.ForeignKey(GraphNodeRun, on_delete=models.CASCADE, null=True, blank=True, related_name="approvals")
-    status = models.CharField(max_length=32, choices=GraphApprovalStatus.choices, default=GraphApprovalStatus.PENDING)
-    reason = models.CharField(max_length=160)
-    payload = models.JSONField(default=dict, blank=True)
-    requested_by = models.CharField(max_length=120, default="graph_runtime")
-    decided_by = models.CharField(max_length=120, blank=True)
-    decided_at = models.DateTimeField(null=True, blank=True)
-    created_at = models.DateTimeField(auto_now_add=True)
-    updated_at = models.DateTimeField(auto_now=True)

graph/native_runtime.py

@@ -1,205 +0,0 @@
-from __future__ import annotations
-
-import time
-from dataclasses import dataclass, field
-from typing import Any
-from uuid import UUID
-
-from asgiref.sync import sync_to_async
-from django.utils import timezone
-
-from control_plane.events.bus import EventBus
-from graph.models import GraphApprovalStatus, GraphEdgeTraversal, GraphNodeRun, GraphNodeRunStatus, GraphRun, GraphRunStatus
-from graph.registry import NodeHandlerRegistry, NodeResult
-from graph.runtime import GraphRuntime
-from graph.spec import ExecutionGraphSpec, outgoing_edges
-
-
-@dataclass
-class GraphExecutionContext:
-    graph_run: GraphRun
-    spec: ExecutionGraphSpec
-    metadata: dict[str, Any] = field(default_factory=dict)
-    signal_payload: dict[str, Any] = field(default_factory=dict)
-
-
-class NativeGraphRuntime(GraphRuntime):
-    def __init__(self, registry: NodeHandlerRegistry, bus: EventBus | None = None) -> None:
-        self.registry = registry
-        self.bus = bus or EventBus()
-
-    async def start(self, project_id: UUID | None = None, *, graph_run: GraphRun | None = None) -> str:
-        if graph_run is None:
-            raise ValueError("NativeGraphRuntime.start requires a persisted GraphRun")
-        if graph_run.status == GraphRunStatus.PENDING:
-            graph_run.status = GraphRunStatus.RUNNING
-            graph_run.started_at = timezone.now()
-            graph_run.current_node = graph_run.current_node or graph_run.execution_graph_version.graph_spec["entry"]
-            graph_run.save(update_fields=["status", "started_at", "current_node", "updated_at"])
-            self.bus.publish("GRAPH_RUN_STARTED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
-        return str(graph_run.id)
-
-    async def pause(self, run_id: str) -> None:
-        graph_run = GraphRun.objects.get(id=run_id)
-        graph_run.status = GraphRunStatus.PAUSED
-        graph_run.save(update_fields=["status", "updated_at"])
-        self.bus.publish("GRAPH_RUN_PAUSED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
-
-    async def resume(self, run_id: str) -> None:
-        graph_run = GraphRun.objects.get(id=run_id)
-        if graph_run.status == GraphRunStatus.PAUSED:
-            graph_run.status = GraphRunStatus.RUNNING
-            graph_run.save(update_fields=["status", "updated_at"])
-            self.bus.publish("GRAPH_RUN_RESUMED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
-        self.run_until_terminal_or_paused(graph_run)
-
-    async def cancel(self, run_id: str) -> None:
-        graph_run = GraphRun.objects.get(id=run_id)
-        graph_run.status = GraphRunStatus.CANCELLED
-        graph_run.completed_at = timezone.now()
-        graph_run.save(update_fields=["status", "completed_at", "updated_at"])
-        self.bus.publish("GRAPH_RUN_CANCELLED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
-
-    async def signal(self, run_id: str, event: dict[str, Any]) -> None:
-        await sync_to_async(self.signal_now)(run_id, event)
-
-    def signal_now(self, run_id: str, event: dict[str, Any]) -> None:
-        graph_run = GraphRun.objects.get(id=run_id)
-        metadata = dict(graph_run.metadata)
-        metadata["last_signal"] = self._bounded(event)
-        if event.get("action") == "approve" and graph_run.status == GraphRunStatus.PAUSED:
-            for approval in graph_run.approvals.filter(status=GraphApprovalStatus.PENDING):
-                approval.status = GraphApprovalStatus.APPROVED
-                approval.decided_by = str(event.get("actor", "system"))
-                approval.decided_at = timezone.now()
-                approval.save(update_fields=["status", "decided_by", "decided_at", "updated_at"])
-            graph_run.status = GraphRunStatus.RUNNING
-        graph_run.metadata = metadata
-        graph_run.save(update_fields=["metadata", "status", "updated_at"])
-        self.bus.publish("GRAPH_RUN_SIGNALED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "signal": self._bounded(event)})
-
-    def run_until_terminal_or_paused(self, graph_run: GraphRun, *, interrupt_after: str | None = None) -> GraphRun:
-        graph_run.refresh_from_db()
-        spec = ExecutionGraphSpec.from_dict(graph_run.execution_graph_version.graph_spec)
-        spec.validate()
-        if graph_run.status == GraphRunStatus.PENDING:
-            graph_run.status = GraphRunStatus.RUNNING
-            graph_run.started_at = timezone.now()
-            graph_run.current_node = graph_run.current_node or spec.entry
-            graph_run.save(update_fields=["status", "started_at", "current_node", "updated_at"])
-            self.bus.publish("GRAPH_RUN_STARTED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
-        while graph_run.status == GraphRunStatus.RUNNING:
-            node_id = graph_run.current_node or spec.entry
-            if node_id in spec.terminal_nodes:
-                self._finish_terminal(graph_run, node_id)
-                break
-            node_spec = spec.nodes[node_id]
-            context = GraphExecutionContext(graph_run=graph_run, spec=spec, metadata=dict(graph_run.metadata))
-            node_run, created = self._node_run(graph_run, node_id, node_spec.node_type)
-            if not created and node_run.status == GraphNodeRunStatus.COMPLETE:
-                result = NodeResult("COMPLETE", str(node_run.output_metadata.get("edge_result", "success")), node_run.output_metadata, node_run.telemetry)
-            else:
-                result = self._execute_node(context, node_run)
-            if interrupt_after == node_id:
-                graph_run.metadata = {**dict(graph_run.metadata), "interrupted_after_node": node_id}
-                graph_run.save(update_fields=["metadata", "updated_at"])
-                graph_run.refresh_from_db()
-                return graph_run
-            if result.status == "PAUSED":
-                graph_run.status = GraphRunStatus.PAUSED
-                graph_run.failure_reason = result.pause_reason
-                graph_run.save(update_fields=["status", "failure_reason", "updated_at"])
-                self.bus.publish("GRAPH_RUN_PAUSED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "reason": result.pause_reason})
-                break
-            if result.status == "FAILED":
-                graph_run.status = GraphRunStatus.FAILED
-                graph_run.completed_at = timezone.now()
-                graph_run.failure_reason = str((result.failure_evidence or {}).get("reason", "node failed"))
-                graph_run.save(update_fields=["status", "completed_at", "failure_reason", "updated_at"])
-                self.bus.publish("GRAPH_RUN_FAILED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "node_id": node_id})
-                break
-            next_node = self._select_next(spec, node_id, result.edge_result)
-            if next_node is None:
-                graph_run.status = GraphRunStatus.FAILED
-                graph_run.completed_at = timezone.now()
-                graph_run.failure_reason = f"No edge from {node_id} for {result.edge_result}"
-                graph_run.save(update_fields=["status", "completed_at", "failure_reason", "updated_at"])
-                self.bus.publish("GRAPH_RUN_FAILED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "reason": graph_run.failure_reason})
-                break
-            GraphEdgeTraversal.objects.create(graph_run=graph_run, source_node=node_id, target_node=next_node, condition=result.edge_result, result=result.edge_result)
-            self.bus.publish("GRAPH_EDGE_TRAVERSED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "source": node_id, "target": next_node, "condition": result.edge_result})
-            graph_run.current_node = next_node
-            graph_run.metadata = self._merge_summary(graph_run.metadata, result)
-            graph_run.metadata["last_node_id"] = node_id
-            graph_run.metadata["last_edge_result"] = result.edge_result
-            graph_run.metadata["last_output"] = self._bounded(result.output_metadata or {})
-            graph_run.metadata.pop("interrupted_after_node", None)
-            graph_run.save(update_fields=["current_node", "metadata", "updated_at"])
-        graph_run.refresh_from_db()
-        return graph_run
-
-    def _node_run(self, graph_run: GraphRun, node_id: str, node_type: str) -> tuple[GraphNodeRun, bool]:
-        last = graph_run.node_runs.filter(node_id=node_id).order_by("-visit_index").first()
-        if last is not None and last.status in {GraphNodeRunStatus.PENDING, GraphNodeRunStatus.RUNNING, GraphNodeRunStatus.PAUSED}:
-            return last, False
-        if last is not None and last.status == GraphNodeRunStatus.COMPLETE and graph_run.current_node == node_id and graph_run.metadata.get("interrupted_after_node") == node_id:
-            return last, False
-        visit_index = 1 if last is None else last.visit_index + 1
-        return GraphNodeRun.objects.create(graph_run=graph_run, node_id=node_id, node_type=node_type, visit_index=visit_index), True
-
-    def _execute_node(self, context: GraphExecutionContext, node_run: GraphNodeRun) -> NodeResult:
-        handler = self.registry.get(node_run.node_type)
-        started = time.monotonic()
-        node_run.status = GraphNodeRunStatus.RUNNING
-        node_run.started_at = timezone.now()
-        node_run.save(update_fields=["status", "started_at", "updated_at"])
-        self.bus.publish("GRAPH_NODE_STARTED", project=context.graph_run.project, task=context.graph_run.task, payload={"graph_run_id": context.graph_run.id, "node_id": node_run.node_id})
-        try:
-            result = handler.run(context)
-        except Exception as exc:
-            result = NodeResult("FAILED", "failure", failure_evidence={"reason": str(exc)})
-        node_run.status = result.status if result.status in GraphNodeRunStatus.values else GraphNodeRunStatus.FAILED
-        node_run.completed_at = timezone.now()
-        output = dict(result.output_metadata or {})
-        output["edge_result"] = result.edge_result
-        node_run.output_metadata = self._bounded(output)
-        telemetry = dict(result.telemetry or {})
-        telemetry["duration_ms"] = int((time.monotonic() - started) * 1000)
-        node_run.telemetry = self._bounded(telemetry)
-        node_run.failure_evidence = self._bounded(result.failure_evidence or {})
-        node_run.save(update_fields=["status", "completed_at", "output_metadata", "telemetry", "failure_evidence", "updated_at"])
-        event = "GRAPH_NODE_COMPLETED" if result.status == "COMPLETE" else "GRAPH_NODE_FAILED"
-        self.bus.publish(event, project=context.graph_run.project, task=context.graph_run.task, payload={"graph_run_id": context.graph_run.id, "node_id": node_run.node_id, "status": result.status})
-        return result
-
-    def _select_next(self, spec: ExecutionGraphSpec, node_id: str, edge_result: str) -> str | None:
-        edges = outgoing_edges(spec, node_id)
-        for edge in edges:
-            if edge.condition == edge_result:
-                return edge.target
-        for edge in edges:
-            if not edge.condition:
-                return edge.target
-        return None
-
-    def _finish_terminal(self, graph_run: GraphRun, node_id: str) -> None:
-        graph_run.status = GraphRunStatus.FAILED if node_id == "fail" else GraphRunStatus.COMPLETE
-        graph_run.completed_at = timezone.now()
-        graph_run.save(update_fields=["status", "completed_at", "updated_at"])
-        event = "GRAPH_RUN_COMPLETED" if graph_run.status == GraphRunStatus.COMPLETE else "GRAPH_RUN_FAILED"
-        self.bus.publish(event, project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "terminal_node": node_id})
-
-    def _merge_summary(self, metadata: dict[str, Any], result: NodeResult) -> dict[str, Any]:
-        summary = dict(metadata)
-        telemetry = dict(summary.get("telemetry", {}))
-        for key, value in (result.telemetry or {}).items():
-            if isinstance(value, int | float):
-                telemetry[key] = telemetry.get(key, 0) + value
-        summary["telemetry"] = telemetry
-        return summary
-
-    def _bounded(self, value: Any) -> Any:
-        text = str(value)
-        if len(text) > 20000:
-            return {"truncated": True, "excerpt": text[:20000]}
-        return value

graph/registry.py

@@ -1,40 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Protocol
-
-
-@dataclass(frozen=True)
-class NodeResult:
-    status: str
-    edge_result: str = "success"
-    output_metadata: dict[str, object] | None = None
-    telemetry: dict[str, object] | None = None
-    failure_evidence: dict[str, object] | None = None
-    pause_reason: str = ""
-
-
-class NodeHandler(Protocol):
-    node_type: str
-    idempotent: bool
-    replay_safe: bool
-    destructive: bool
-
-    def run(self, context: object) -> NodeResult: ...
-
-
-class NodeHandlerRegistry:
-    def __init__(self) -> None:
-        self._handlers: dict[str, NodeHandler] = {}
-
-    def register(self, handler: NodeHandler) -> None:
-        self._handlers[handler.node_type] = handler
-
-    def get(self, node_type: str) -> NodeHandler:
-        try:
-            return self._handlers[node_type]
-        except KeyError as exc:
-            raise KeyError(f"No graph node handler registered for {node_type}") from exc
-
-    def has(self, node_type: str) -> bool:
-        return node_type in self._handlers

graph/runtime.py

@@ -9,7 +9,7 @@ class GraphRuntime(ABC):
     """Execution runtime boundary. LangGraph must stay behind this interface."""
 
     @abstractmethod
-    async def start(self, project_id: UUID | None = None, **kwargs: Any) -> str: ...
+    async def start(self, project_id: UUID) -> str: ...
 
     @abstractmethod
     async def pause(self, run_id: str) -> None: ...

graph/spec.py

@@ -1,91 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any
-
-
-@dataclass(frozen=True)
-class GraphNodeSpec:
-    node_id: str
-    node_type: str
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {"id": self.node_id, "type": self.node_type, "metadata": self.metadata}
-
-
-@dataclass(frozen=True)
-class GraphEdgeSpec:
-    source: str
-    target: str
-    condition: str = ""
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {"source": self.source, "target": self.target, "condition": self.condition, "metadata": self.metadata}
-
-
-@dataclass(frozen=True)
-class ExecutionGraphSpec:
-    name: str
-    version: int
-    graph_type: str
-    entry: str
-    nodes: dict[str, GraphNodeSpec]
-    edges: list[GraphEdgeSpec]
-    terminal_nodes: list[str]
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "name": self.name,
-            "version": self.version,
-            "graph_type": self.graph_type,
-            "entry": self.entry,
-            "nodes": {node_id: node.to_dict() for node_id, node in self.nodes.items()},
-            "edges": [edge.to_dict() for edge in self.edges],
-            "terminal_nodes": self.terminal_nodes,
-            "metadata": self.metadata,
-        }
-
-    @classmethod
-    def from_dict(cls, payload: dict[str, Any]) -> "ExecutionGraphSpec":
-        nodes = {
-            node_id: GraphNodeSpec(node_id=str(raw["id"]), node_type=str(raw["type"]), metadata=dict(raw.get("metadata", {})))
-            for node_id, raw in dict(payload["nodes"]).items()
-        }
-        edges = [
-            GraphEdgeSpec(
-                source=str(raw["source"]),
-                target=str(raw["target"]),
-                condition=str(raw.get("condition", "")),
-                metadata=dict(raw.get("metadata", {})),
-            )
-            for raw in list(payload["edges"])
-        ]
-        return cls(
-            name=str(payload["name"]),
-            version=int(payload["version"]),
-            graph_type=str(payload["graph_type"]),
-            entry=str(payload["entry"]),
-            nodes=nodes,
-            edges=edges,
-            terminal_nodes=[str(item) for item in payload.get("terminal_nodes", [])],
-            metadata=dict(payload.get("metadata", {})),
-        )
-
-    def validate(self) -> None:
-        if self.entry not in self.nodes:
-            raise ValueError(f"Graph entry node does not exist: {self.entry}")
-        for terminal in self.terminal_nodes:
-            if terminal not in self.nodes:
-                raise ValueError(f"Terminal node does not exist: {terminal}")
-        for edge in self.edges:
-            if edge.source not in self.nodes:
-                raise ValueError(f"Edge source does not exist: {edge.source}")
-            if edge.target not in self.nodes:
-                raise ValueError(f"Edge target does not exist: {edge.target}")
-
-
-def outgoing_edges(spec: ExecutionGraphSpec, node_id: str) -> list[GraphEdgeSpec]:
-    return [edge for edge in spec.edges if edge.source == node_id]

graph/task_execution.py

@@ -1,54 +0,0 @@
-from __future__ import annotations
-
-from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
-
-
-TASK_EXECUTION_GRAPH_NAME = "task_execution"
-TASK_EXECUTION_GRAPH_VERSION = 1
-
-
-def task_execution_graph_v1() -> ExecutionGraphSpec:
-    nodes = {
-        "claim_task": GraphNodeSpec("claim_task", "claim_task", {"idempotent": False, "replay_safe": False}),
-        "prepare_worktree": GraphNodeSpec("prepare_worktree", "prepare_worktree", {"idempotent": True, "replay_safe": True}),
-        "build_context": GraphNodeSpec("build_context", "build_context", {"idempotent": True, "replay_safe": True}),
-        "coder": GraphNodeSpec("coder", "coder", {"idempotent": False, "replay_safe": "checkpointed", "contains_internal_tool_loop": True}),
-        "run_tests": GraphNodeSpec("run_tests", "run_tests", {"idempotent": True, "replay_safe": "checkpointed"}),
-        "review": GraphNodeSpec("review", "review", {"idempotent": True, "replay_safe": "checkpointed"}),
-        "judge": GraphNodeSpec("judge", "judge", {"idempotent": True, "replay_safe": "checkpointed"}),
-        "commit": GraphNodeSpec("commit", "commit", {"idempotent": False, "replay_safe": "guarded", "destructive": True}),
-        "retry_or_fail": GraphNodeSpec("retry_or_fail", "retry_or_fail", {"idempotent": False, "replay_safe": "checkpointed"}),
-        "cleanup": GraphNodeSpec("cleanup", "cleanup", {"idempotent": True, "replay_safe": True}),
-        "complete": GraphNodeSpec("complete", "complete", {"terminal": True}),
-        "fail": GraphNodeSpec("fail", "fail", {"terminal": True}),
-    }
-    edges = [
-        GraphEdgeSpec("claim_task", "prepare_worktree", "success"),
-        GraphEdgeSpec("prepare_worktree", "build_context", "success"),
-        GraphEdgeSpec("build_context", "coder", "success"),
-        GraphEdgeSpec("coder", "run_tests", "success"),
-        GraphEdgeSpec("coder", "retry_or_fail", "failure"),
-        GraphEdgeSpec("run_tests", "review", "complete"),
-        GraphEdgeSpec("review", "judge", "PASS"),
-        GraphEdgeSpec("review", "retry_or_fail", "REWORK_REQUIRED"),
-        GraphEdgeSpec("review", "retry_or_fail", "REJECTED"),
-        GraphEdgeSpec("judge", "commit", "PASS"),
-        GraphEdgeSpec("judge", "retry_or_fail", "FAIL"),
-        GraphEdgeSpec("commit", "cleanup", "success"),
-        GraphEdgeSpec("cleanup", "complete", "success"),
-        GraphEdgeSpec("retry_or_fail", "build_context", "retry_available"),
-        GraphEdgeSpec("retry_or_fail", "cleanup", "retry_exhausted"),
-        GraphEdgeSpec("cleanup", "fail", "failed"),
-    ]
-    spec = ExecutionGraphSpec(
-        name=TASK_EXECUTION_GRAPH_NAME,
-        version=TASK_EXECUTION_GRAPH_VERSION,
-        graph_type="TASK_EXECUTION",
-        entry="claim_task",
-        nodes=nodes,
-        edges=edges,
-        terminal_nodes=["complete", "fail"],
-        metadata={"description": "Task execution graph v1 mirrors AutonomousTaskLoop semantics."},
-    )
-    spec.validate()
-    return spec

graph/task_nodes.py

@@ -1,356 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from agents.coder import Coder
-from agents.judge import Judge
-from agents.reviewer import Reviewer
-from control_plane.agents.models import AgentRole, AgentVersion
-from control_plane.events.bus import EventBus
-from control_plane.events.models import EventType
-from control_plane.projects.models import CommitRecord, Task, TaskAttempt, TaskStatus, Worktree
-from control_plane.verification.models import TestRun, VerificationResult
-from graph.native_runtime import GraphExecutionContext
-from graph.registry import NodeHandlerRegistry, NodeResult
-from knowledge.context_builder import WorkerContextBuilder
-from model_router.router import ModelRouter
-from tools.capabilities import Capability
-from tools.runtime import WorktreeTools
-from tools.test_runner import DeterministicTestRunner
-from workspace.worktrees import WorktreeManager
-
-
-class TaskExecutionServices:
-    def __init__(
-        self,
-        router: ModelRouter,
-        *,
-        bus: EventBus | None = None,
-        test_command: list[str] | None = None,
-    ) -> None:
-        self.router = router
-        self.bus = bus or EventBus()
-        self.context_builder = WorkerContextBuilder()
-        self.worktrees = WorktreeManager()
-        self.coder = Coder(router)
-        self.reviewer = Reviewer()
-        self.judge = Judge()
-        self.tests = DeterministicTestRunner()
-        self.test_command = test_command or ["python", "-m", "pytest"]
-
-    def champion(self, role: AgentRole) -> AgentVersion:
-        return AgentVersion.objects.select_related("agent").get(agent__role=role, promotion_status="CHAMPION")
-
-    def tools(self, worktree: Worktree) -> WorktreeTools:
-        return WorktreeTools(
-            Path(worktree.worktree_path),
-            {
-                Capability.READ_REPOSITORY,
-                Capability.INVESTIGATE_WORKTREE,
-                Capability.WRITE_WORKTREE,
-                Capability.RUN_TESTS,
-                Capability.COMMIT_CHANGES,
-            },
-        )
-
-    def worktree(self, task: Task) -> Worktree:
-        try:
-            return task.worktree
-        except Worktree.DoesNotExist:
-            if not task.project.repository_path:
-                raise RuntimeError("Task project has no repository_path")
-            return self.worktrees.create_for_task(task, Path(task.project.repository_path))
-
-
-class TaskNode:
-    idempotent = True
-    replay_safe = "checkpointed"
-    destructive = False
-
-    def __init__(self, services: TaskExecutionServices, node_type: str) -> None:
-        self.services = services
-        self.node_type = node_type
-
-    def task(self, context: GraphExecutionContext) -> Task:
-        if context.graph_run.task_id is None:
-            raise RuntimeError("Task execution graph run requires task")
-        return Task.objects.select_related("project", "milestone", "feature").get(id=context.graph_run.task_id)
-
-    def metadata(self, context: GraphExecutionContext) -> dict[str, object]:
-        return dict(context.graph_run.metadata)
-
-    def save_metadata(self, context: GraphExecutionContext, metadata: dict[str, object]) -> None:
-        context.graph_run.metadata = metadata
-        context.graph_run.save(update_fields=["metadata", "updated_at"])
-
-
-class ClaimTaskNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "claim_task")
-        self.idempotent = True
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        return NodeResult("COMPLETE", "success", {"task_id": str(task.id), "task_status": task.status})
-
-
-class PrepareWorktreeNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "prepare_worktree")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        worktree = self.services.worktree(task)
-        metadata = self.metadata(context)
-        metadata["worktree_id"] = str(worktree.id)
-        metadata["worktree_path"] = worktree.worktree_path
-        self.save_metadata(context, metadata)
-        return NodeResult("COMPLETE", "success", {"worktree_id": str(worktree.id), "worktree_path": worktree.worktree_path})
-
-
-class BuildContextNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "build_context")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        worktree = self.services.worktree(task)
-        metadata = self.metadata(context)
-        attempt_id = metadata.get("current_attempt_id")
-        if attempt_id:
-            attempt = TaskAttempt.objects.get(id=attempt_id)
-            return NodeResult("COMPLETE", "success", {"attempt_id": str(attempt.id), "attempt_number": attempt.attempt_number})
-        coder_version = self.services.champion(AgentRole.CODER)
-        attempt = TaskAttempt.objects.create(task=task, attempt_number=task.retry_count + 1, coder=coder_version, status="RUNNING")
-        task_context = self.services.context_builder.build_for_task(task, Path(worktree.worktree_path))
-        task_context["previous_attempts"] = list(
-            task.attempts.exclude(id=attempt.id).order_by("attempt_number").values("attempt_number", "status", "coder_result", "review_findings", "judge_findings")
-        )
-        attempt.context_snapshot = task_context
-        attempt.save(update_fields=["context_snapshot", "updated_at"])
-        metadata["current_attempt_id"] = str(attempt.id)
-        self.save_metadata(context, metadata)
-        context.graph_run.task_attempt = attempt
-        context.graph_run.save(update_fields=["task_attempt", "updated_at"])
-        return NodeResult("COMPLETE", "success", {"attempt_id": str(attempt.id), "attempt_number": attempt.attempt_number})
-
-
-class CoderNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "coder")
-        self.idempotent = False
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        worktree = self.services.worktree(task)
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        coder_version = attempt.coder
-        result = self.services.coder.execute(attempt.context_snapshot, self.services.tools(worktree), project=task.project, agent_version=coder_version)
-        attempt.coder_result = {"status": result.status, "summary": result.summary, "changed_files": result.changed_files, "metadata": result.metadata}
-        attempt.save(update_fields=["coder_result", "updated_at"])
-        metadata["last_failure_reason"] = "coder_failed"
-        metadata["last_failure_findings"] = [result.summary]
-        self.save_metadata(context, metadata)
-        return NodeResult("COMPLETE", "success" if result.status == "COMPLETE" else "failure", {"coder_status": result.status, "summary": result.summary}, result.metadata.get("telemetry", {}) if isinstance(result.metadata, dict) else {})
-
-
-class RunTestsNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "run_tests")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        worktree = self.services.worktree(task)
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        test_run = self.services.tests.run(task.project, task, Path(worktree.worktree_path), self.services.test_command)
-        metadata["test_run_id"] = str(test_run.id)
-        self.save_metadata(context, metadata)
-        if test_run.status != "PASS":
-            self._attach_test_failure_evidence(attempt, test_run)
-            self.services.bus.publish(EventType.TEST_FAILED, project=task.project, task=task, payload={"test_run_id": str(test_run.id)})
-        return NodeResult("COMPLETE", "complete", {"test_run_id": str(test_run.id), "test_status": test_run.status}, {"test_status": test_run.status})
-
-    def _attach_test_failure_evidence(self, attempt: TaskAttempt, test_run: TestRun) -> None:
-        content = test_run.output_artifact.content if test_run.output_artifact else {}
-        stdout = str(content.get("stdout", "")) if isinstance(content, dict) else ""
-        stderr = str(content.get("stderr", "")) if isinstance(content, dict) else ""
-        coder_result = dict(attempt.coder_result or {})
-        attempt_metadata = dict(coder_result.get("metadata", {})) if isinstance(coder_result.get("metadata", {}), dict) else {}
-        attempt_metadata["test_failure_evidence"] = {"test_run_id": str(test_run.id), "status": test_run.status, "stdout_excerpt": stdout[-12000:], "stderr_excerpt": stderr[-4000:]}
-        coder_result["metadata"] = attempt_metadata
-        attempt.coder_result = coder_result
-        attempt.save(update_fields=["coder_result", "updated_at"])
-
-
-class ReviewNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "review")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        reviewer_version = self.services.champion(AgentRole.REVIEWER)
-        worktree = self.services.worktree(task)
-        tools = self.services.tools(worktree)
-        tools.git(["add", "-N", "."])
-        diff = tools.diff()
-        test_run = TestRun.objects.get(id=metadata["test_run_id"])
-        review = self.services.reviewer.review(task, reviewer_version, diff, test_run.status)
-        attempt.review_findings = review.findings
-        attempt.save(update_fields=["review_findings", "updated_at"])
-        metadata["review_id"] = str(review.id)
-        metadata["last_failure_reason"] = "review_failed"
-        metadata["last_failure_findings"] = review.findings
-        self.save_metadata(context, metadata)
-        if review.status != "PASS":
-            self.services.bus.publish(EventType.REVIEW_FAILED, project=task.project, task=task, payload={"review_id": str(review.id), "findings": review.findings})
-        return NodeResult("COMPLETE", review.status, {"review_id": str(review.id), "review_status": review.status, "findings": review.findings}, {"review_status": review.status})
-
-
-class JudgeNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "judge")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        judge_version = self.services.champion(AgentRole.PROJECT_JUDGE)
-        test_run = TestRun.objects.get(id=metadata["test_run_id"])
-        worktree = self.services.worktree(task)
-        tools = self.services.tools(worktree)
-        tools.git(["add", "-N", "."])
-        diff = tools.diff()
-        verification = self.services.judge.judge(task.project, task, judge_version, diff, test_run.status)
-        attempt.judge_findings = verification.evidence
-        attempt.save(update_fields=["judge_findings", "updated_at"])
-        metadata["verification_id"] = str(verification.id)
-        metadata["last_failure_reason"] = "judge_failed"
-        metadata["last_failure_findings"] = verification.evidence
-        self.save_metadata(context, metadata)
-        edge = "PASS" if verification.result == VerificationResult.PASS else "FAIL"
-        return NodeResult("COMPLETE", edge, {"verification_id": str(verification.id), "result": verification.result, "evidence": verification.evidence}, {"judge_result": verification.result})
-
-
-class RetryOrFailNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "retry_or_fail")
-        self.idempotent = False
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        effective_max_retries = min(task.max_retries, 2)
-        attempt.status = "REWORK_REQUIRED" if task.retry_count < effective_max_retries else "FAILED"
-        attempt.save(update_fields=["status", "updated_at"])
-        task.retry_count += 1
-        reason = str(metadata.get("last_failure_reason", "task_failed"))
-        findings = metadata.get("last_failure_findings", [])
-        classification = self._classify_failure(reason, findings)
-        metadata.pop("current_attempt_id", None)
-        self.save_metadata(context, metadata)
-        if task.retry_count <= effective_max_retries:
-            task.status = TaskStatus.RUNNING
-            task.save(update_fields=["retry_count", "status", "updated_at"])
-            self.services.bus.publish(EventType.TASK_FAILED, project=task.project, task=task, payload={"reason": reason, "will_retry": True, "classification": classification, "findings": findings})
-            return NodeResult("COMPLETE", "retry_available", {"retry_count": task.retry_count, "classification": classification})
-        task.status = TaskStatus.FAILED
-        task.save(update_fields=["retry_count", "status", "updated_at"])
-        self.services.bus.publish("TASK_RETRY_EXHAUSTED", project=task.project, task=task, payload={"reason": reason, "classification": classification, "findings": findings})
-        self.services.bus.publish(EventType.TASK_FAILED, project=task.project, task=task, payload={"reason": reason, "will_retry": False, "classification": classification, "findings": findings})
-        return NodeResult("COMPLETE", "retry_exhausted", {"retry_count": task.retry_count, "classification": classification})
-
-    def _classify_failure(self, reason: str, findings: object) -> str:
-        text = f"{reason} {findings}".lower()
-        if "unsupported operation" in text or "missing capability" in text:
-            return "missing_capability"
-        if "context" in text or "migration" in text:
-            return "context_problem"
-        if "timeout" in text or "provider" in text:
-            return "environment_problem"
-        if "malformed json" in text or "model" in text:
-            return "model_problem"
-        if "ambiguous" in text:
-            return "intent_ambiguity"
-        if reason == "review_failed" or reason == "judge_failed":
-            return "replan"
-        return "split_task"
-
-
-class CommitNode(TaskNode):
-    destructive = True
-
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "commit")
-        self.idempotent = False
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        existing = CommitRecord.objects.filter(task=task).first()
-        if existing is not None:
-            return NodeResult("COMPLETE", "success", {"commit_id": str(existing.id), "sha": existing.sha, "deduplicated": True})
-        metadata = self.metadata(context)
-        attempt = TaskAttempt.objects.get(id=metadata["current_attempt_id"])
-        worktree = self.services.worktree(task)
-        test_run = TestRun.objects.get(id=metadata["test_run_id"])
-        sha = self.services.tools(worktree).commit_all(f"Artifex task: {task.goal[:80]}")
-        commit = CommitRecord.objects.create(
-            project=task.project,
-            task=task,
-            worktree=worktree,
-            coder=attempt.coder,
-            reviewer=self.services.champion(AgentRole.REVIEWER),
-            judge=self.services.champion(AgentRole.PROJECT_JUDGE),
-            test_run=test_run,
-            review_id=metadata.get("review_id"),
-            verification_id=metadata.get("verification_id"),
-            graph_run=context.graph_run,
-            sha=sha,
-            branch_name=worktree.branch_name,
-            message=f"Artifex task: {task.goal[:80]}",
-        )
-        attempt.status = "COMPLETE"
-        attempt.save(update_fields=["status", "updated_at"])
-        task.status = TaskStatus.COMPLETE
-        task.save(update_fields=["status", "updated_at"])
-        self.services.bus.publish(EventType.COMMIT_CREATED, project=task.project, task=task, payload={"commit_id": str(commit.id), "sha": sha})
-        self.services.bus.publish(EventType.TASK_COMPLETED, project=task.project, task=task, payload={"task_id": str(task.id)})
-        metadata["commit_id"] = str(commit.id)
-        metadata["commit_sha"] = sha
-        self.save_metadata(context, metadata)
-        return NodeResult("COMPLETE", "success", {"commit_id": str(commit.id), "sha": sha})
-
-
-class CleanupNode(TaskNode):
-    def __init__(self, services: TaskExecutionServices) -> None:
-        super().__init__(services, "cleanup")
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        task = self.task(context)
-        if task.status == TaskStatus.COMPLETE:
-            self.services.worktrees.validate_clean_worktree(task.worktree)
-            self.services.worktrees.cleanup(task.worktree)
-            return NodeResult("COMPLETE", "success", {"cleaned": True})
-        return NodeResult("COMPLETE", "failed", {"cleaned": False})
-
-
-def task_execution_registry(services: TaskExecutionServices) -> NodeHandlerRegistry:
-    registry = NodeHandlerRegistry()
-    for handler in [
-        ClaimTaskNode(services),
-        PrepareWorktreeNode(services),
-        BuildContextNode(services),
-        CoderNode(services),
-        RunTestsNode(services),
-        ReviewNode(services),
-        JudgeNode(services),
-        CommitNode(services),
-        RetryOrFailNode(services),
-        CleanupNode(services),
-    ]:
-        registry.register(handler)
-    return registry

pyproject.toml

@@ -6,7 +6,6 @@ requires-python = ">=3.12"
 dependencies = [
     "django>=5.1,<6.0",
     "psycopg[binary]>=3.2,<4.0",
-    "langgraph>=0.2,<0.3",
     "structlog>=24.4,<25.0",
 ]
 

runtime_loop/autonomous_task_loop.py

@@ -10,11 +10,7 @@ from control_plane.events.bus import EventBus
 from control_plane.events.models import EventType
 from control_plane.projects.models import CommitRecord, Task, TaskAttempt, TaskStatus, Worktree
 from control_plane.verification.models import VerificationResult
-from graph.bootstrap import champion_task_execution_graph_v1
-from graph.models import GraphRun
-from graph.native_runtime import NativeGraphRuntime
 from graph.scheduler import TaskScheduler
-from graph.task_nodes import TaskExecutionServices, task_execution_registry
 from knowledge.context_builder import WorkerContextBuilder
 from model_router.router import ModelRouter
 from tools.capabilities import Capability
@@ -49,19 +45,6 @@ class AutonomousTaskLoop:
         return task
 
     def _execute_task(self, task: Task, test_command: list[str]) -> None:
-        graph_version = champion_task_execution_graph_v1()
-        graph_run = GraphRun.objects.create(
-            execution_graph_version=graph_version,
-            project=task.project,
-            milestone=task.milestone,
-            feature=task.feature,
-            task=task,
-            current_node=graph_version.graph_spec["entry"],
-        )
-        services = TaskExecutionServices(self.router, bus=self.bus, test_command=test_command)
-        NativeGraphRuntime(task_execution_registry(services), bus=self.bus).run_until_terminal_or_paused(graph_run)
-        return
-
         coder_version = self._champion(AgentRole.CODER)
         reviewer_version = self._champion(AgentRole.REVIEWER)
         judge_version = self._champion(AgentRole.PROJECT_JUDGE)

tests/test_execution_graph_phase_a.py

@@ -1,92 +0,0 @@
-from __future__ import annotations
-
-import pytest
-
-from control_plane.agents.models import ProgenySignal
-from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphRun, GraphRunStatus
-from graph.registry import NodeHandlerRegistry, NodeResult
-from graph.spec import ExecutionGraphSpec, outgoing_edges
-from graph.task_execution import task_execution_graph_v1
-
-
-class FixtureNode:
-    node_type = "fixture"
-    idempotent = True
-    replay_safe = True
-    destructive = False
-
-    def run(self, context: object) -> NodeResult:
-        return NodeResult("COMPLETE", "success", {"ok": True})
-
-
-def test_task_execution_graph_v1_is_serializable_and_valid() -> None:
-    spec = task_execution_graph_v1()
-
-    payload = spec.to_dict()
-    restored = ExecutionGraphSpec.from_dict(payload)
-    restored.validate()
-
-    assert restored.name == "task_execution"
-    assert restored.version == 1
-    assert restored.entry == "claim_task"
-    assert restored.nodes["coder"].metadata["contains_internal_tool_loop"] is True
-    assert {edge.condition for edge in outgoing_edges(restored, "review")} == {"PASS", "REWORK_REQUIRED", "REJECTED"}
-    assert {edge.condition for edge in outgoing_edges(restored, "retry_or_fail")} == {"retry_available", "retry_exhausted"}
-
-
-def test_node_registry_resolves_handlers_without_arbitrary_execution() -> None:
-    registry = NodeHandlerRegistry()
-    registry.register(FixtureNode())
-
-    assert registry.get("fixture").run(object()).output_metadata == {"ok": True}
-    with pytest.raises(KeyError):
-        registry.get("missing")
-
-
-def test_graph_models_persist_versioned_run_state() -> None:
-    spec = task_execution_graph_v1()
-    definition = ExecutionGraphDefinition.objects.create(
-        name=spec.name,
-        graph_type=spec.graph_type,
-        description="Task execution graph",
-    )
-    version = ExecutionGraphVersion.objects.create(
-        graph=definition,
-        version=spec.version,
-        status=ExecutionGraphVersionStatus.CHAMPION,
-        graph_spec=spec.to_dict(),
-        metadata={"immutable_after_use": True},
-    )
-
-    run = GraphRun.objects.create(
-        execution_graph_version=version,
-        status=GraphRunStatus.PENDING,
-        current_node=spec.entry,
-        metadata={"node_count": len(spec.nodes)},
-    )
-
-    assert run.execution_graph_version.graph.name == "task_execution"
-    assert run.execution_graph_version.status == ExecutionGraphVersionStatus.CHAMPION
-    assert run.metadata["node_count"] == 12
-
-
-def test_progeny_signal_can_reference_graph_lineage() -> None:
-    spec = task_execution_graph_v1()
-    definition = ExecutionGraphDefinition.objects.create(name="lineage_graph", graph_type=spec.graph_type)
-    version = ExecutionGraphVersion.objects.create(graph=definition, version=1, status=ExecutionGraphVersionStatus.CHAMPION, graph_spec=spec.to_dict())
-    run = GraphRun.objects.create(execution_graph_version=version, current_node=spec.entry)
-    node_run = run.node_runs.create(node_id="coder", node_type="coder", visit_index=1)
-
-    signal = ProgenySignal.objects.create(
-        source="runtime",
-        severity="high",
-        failure_category="MODEL_OUTPUT_INVALID",
-        summary="Malformed model JSON",
-        graph_run=run,
-        graph_node_run=node_run,
-        execution_graph_version=version,
-    )
-
-    assert signal.graph_run == run
-    assert signal.graph_node_run == node_run
-    assert signal.execution_graph_version == version

tests/test_graph_pause_subgraph_inspection.py

@@ -1,102 +0,0 @@
-from __future__ import annotations
-
-from graph.inspection import graph_run_inspection
-from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphApproval, GraphApprovalStatus, GraphRun, GraphRunStatus
-from graph.native_runtime import GraphExecutionContext, NativeGraphRuntime
-from graph.registry import NodeHandlerRegistry, NodeResult
-from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
-
-
-def persist_spec(spec: ExecutionGraphSpec) -> GraphRun:
-    definition = ExecutionGraphDefinition.objects.create(name=spec.name, graph_type=spec.graph_type)
-    version = ExecutionGraphVersion.objects.create(graph=definition, version=spec.version, status=ExecutionGraphVersionStatus.CHAMPION, graph_spec=spec.to_dict())
-    return GraphRun.objects.create(execution_graph_version=version, current_node=spec.entry)
-
-
-class ApprovalGate:
-    node_type = "approval_gate"
-    idempotent = True
-    replay_safe = "checkpointed"
-    destructive = False
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        node_run = context.graph_run.node_runs.filter(node_id=context.graph_run.current_node).order_by("-visit_index").first()
-        if GraphApproval.objects.filter(graph_run=context.graph_run, status=GraphApprovalStatus.APPROVED).exists():
-            return NodeResult("COMPLETE", "approved")
-        GraphApproval.objects.get_or_create(graph_run=context.graph_run, node_run=node_run, reason="AWAITING_APPROVAL")
-        return NodeResult("PAUSED", "awaiting", pause_reason="AWAITING_APPROVAL")
-
-
-class DoneNode:
-    node_type = "done_node"
-    idempotent = True
-    replay_safe = True
-    destructive = False
-
-    def run(self, context: GraphExecutionContext) -> NodeResult:
-        return NodeResult("COMPLETE", "success")
-
-
-def test_human_approval_pause_signal_resume() -> None:
-    spec = ExecutionGraphSpec(
-        name="approval_fixture",
-        version=1,
-        graph_type="FIXTURE",
-        entry="approval",
-        nodes={"approval": GraphNodeSpec("approval", "approval_gate"), "done": GraphNodeSpec("done", "done_node")},
-        edges=[GraphEdgeSpec("approval", "done", "approved")],
-        terminal_nodes=["done"],
-    )
-    graph_run = persist_spec(spec)
-    registry = NodeHandlerRegistry()
-    registry.register(ApprovalGate())
-    registry.register(DoneNode())
-    runtime = NativeGraphRuntime(registry)
-
-    runtime.run_until_terminal_or_paused(graph_run)
-    graph_run.refresh_from_db()
-    assert graph_run.status == GraphRunStatus.PAUSED
-    assert graph_run.approvals.get().status == GraphApprovalStatus.PENDING
-
-    runtime.signal_now(str(graph_run.id), {"action": "approve", "actor": "tester"})
-    runtime.run_until_terminal_or_paused(graph_run)
-    graph_run.refresh_from_db()
-    assert graph_run.status == GraphRunStatus.COMPLETE
-    assert graph_run.approvals.get().status == GraphApprovalStatus.APPROVED
-
-
-def test_subgraph_can_be_represented_in_runtime_neutral_spec() -> None:
-    spec = ExecutionGraphSpec(
-        name="subgraph_fixture",
-        version=1,
-        graph_type="FIXTURE",
-        entry="parent",
-        nodes={
-            "parent": GraphNodeSpec("parent", "subgraph", {"subgraph": {"name": "child_graph", "version": 1}}),
-            "done": GraphNodeSpec("done", "done_node"),
-        },
-        edges=[GraphEdgeSpec("parent", "done", "success")],
-        terminal_nodes=["done"],
-    )
-
-    restored = ExecutionGraphSpec.from_dict(spec.to_dict())
-
-    assert restored.nodes["parent"].metadata["subgraph"]["name"] == "child_graph"
-
-
-def test_graph_run_inspection_exposes_ui_ready_shape() -> None:
-    spec = ExecutionGraphSpec(
-        name="inspect_fixture",
-        version=1,
-        graph_type="FIXTURE",
-        entry="done",
-        nodes={"done": GraphNodeSpec("done", "done_node")},
-        edges=[],
-        terminal_nodes=["done"],
-    )
-    graph_run = persist_spec(spec)
-    inspection = graph_run_inspection(graph_run)
-
-    assert inspection["graph"] == "inspect_fixture"
-    assert inspection["current_node"] == "done"
-    assert inspection["nodes"] == [{"id": "done", "type": "done_node", "status": "PENDING", "visit_index": 0, "duration_ms": None, "failure": {}, "metadata": {}}]

tests/test_langgraph_runtime_adapter.py

@@ -1,17 +0,0 @@
-from __future__ import annotations
-
-import importlib.util
-
-import pytest
-
-from graph.langgraph_runtime import LangGraphRuntime
-
-
-def test_langgraph_runtime_reports_missing_dependency() -> None:
-    if importlib.util.find_spec("langgraph") is not None:
-        pytest.skip("langgraph is installed; adapter execution is covered by integration parity tests")
-
-    runtime = LangGraphRuntime()
-
-    with pytest.raises(RuntimeError, match="langgraph package|NodeHandlerRegistry"):
-        runtime.run_until_terminal_or_paused(None)  # type: ignore[arg-type]

tests/test_m2_autonomous_loop.py

@@ -85,8 +85,6 @@ def assert_accepted_trace(task: Task) -> CommitRecord:
     assert review.status == "PASS"
     assert verification.result == VerificationResult.PASS
     assert commit.sha
-    assert commit.graph_run_id is not None
-    assert commit.graph_run.execution_graph_version.graph.name == "task_execution"
     assert commit.coder.agent.role == "CODER"
     assert commit.reviewer.agent.role == "REVIEWER"
     assert commit.judge.agent.role == "PROJECT_JUDGE"

tests/test_native_graph_runtime.py

@@ -1,114 +0,0 @@
-from __future__ import annotations
-
-from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphEdgeTraversal, GraphRun, GraphRunStatus
-from graph.native_runtime import NativeGraphRuntime
-from graph.registry import NodeHandlerRegistry, NodeResult
-from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
-
-
-class FixedNode:
-    idempotent = True
-    replay_safe = True
-    destructive = False
-
-    def __init__(self, node_type: str, edge_result: str = "success") -> None:
-        self.node_type = node_type
-        self.edge_result = edge_result
-        self.calls = 0
-
-    def run(self, context: object) -> NodeResult:
-        self.calls += 1
-        return NodeResult("COMPLETE", self.edge_result, {"calls": self.calls}, {"model_requests": 1})
-
-
-def persist_spec(spec: ExecutionGraphSpec) -> GraphRun:
-    definition = ExecutionGraphDefinition.objects.create(name=spec.name, graph_type=spec.graph_type)
-    version = ExecutionGraphVersion.objects.create(graph=definition, version=spec.version, status=ExecutionGraphVersionStatus.CHAMPION, graph_spec=spec.to_dict())
-    return GraphRun.objects.create(execution_graph_version=version, current_node=spec.entry)
-
-
-def test_native_graph_runtime_executes_conditional_edges_and_persists_history() -> None:
-    spec = ExecutionGraphSpec(
-        name="fixture_graph",
-        version=1,
-        graph_type="FIXTURE",
-        entry="start",
-        nodes={
-            "start": GraphNodeSpec("start", "start"),
-            "success": GraphNodeSpec("success", "success"),
-            "fail": GraphNodeSpec("fail", "fail"),
-        },
-        edges=[GraphEdgeSpec("start", "success", "ok"), GraphEdgeSpec("start", "fail", "bad")],
-        terminal_nodes=["success", "fail"],
-    )
-    graph_run = persist_spec(spec)
-    registry = NodeHandlerRegistry()
-    start = FixedNode("start", "ok")
-    registry.register(start)
-
-    result = NativeGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
-
-    assert result.status == GraphRunStatus.COMPLETE
-    assert result.current_node == "success"
-    assert start.calls == 1
-    assert list(GraphEdgeTraversal.objects.filter(graph_run=graph_run).values_list("source_node", "target_node", "condition")) == [("start", "success", "ok")]
-    assert graph_run.node_runs.get(node_id="start").telemetry["duration_ms"] >= 0
-
-
-def test_native_graph_runtime_resume_does_not_repeat_completed_current_node() -> None:
-    spec = ExecutionGraphSpec(
-        name="resume_graph",
-        version=1,
-        graph_type="FIXTURE",
-        entry="first",
-        nodes={
-            "first": GraphNodeSpec("first", "first"),
-            "second": GraphNodeSpec("second", "second"),
-            "done": GraphNodeSpec("done", "done"),
-        },
-        edges=[GraphEdgeSpec("first", "second", "success"), GraphEdgeSpec("second", "done", "success")],
-        terminal_nodes=["done"],
-    )
-    graph_run = persist_spec(spec)
-    registry = NodeHandlerRegistry()
-    first = FixedNode("first")
-    second = FixedNode("second")
-    registry.register(first)
-    registry.register(second)
-    runtime = NativeGraphRuntime(registry)
-
-    runtime.run_until_terminal_or_paused(graph_run, interrupt_after="first")
-    graph_run.refresh_from_db()
-    assert graph_run.current_node == "first"
-
-    runtime.run_until_terminal_or_paused(graph_run)
-
-    graph_run.refresh_from_db()
-    assert graph_run.status == GraphRunStatus.COMPLETE
-    assert first.calls == 1
-    assert second.calls == 1
-
-
-def test_native_graph_runtime_records_paused_runs() -> None:
-    class PauseNode(FixedNode):
-        def run(self, context: object) -> NodeResult:
-            self.calls += 1
-            return NodeResult("PAUSED", "awaiting", pause_reason="AWAITING_APPROVAL")
-
-    spec = ExecutionGraphSpec(
-        name="pause_graph",
-        version=1,
-        graph_type="FIXTURE",
-        entry="approval",
-        nodes={"approval": GraphNodeSpec("approval", "approval"), "done": GraphNodeSpec("done", "done")},
-        edges=[GraphEdgeSpec("approval", "done", "approved")],
-        terminal_nodes=["done"],
-    )
-    graph_run = persist_spec(spec)
-    registry = NodeHandlerRegistry()
-    registry.register(PauseNode("approval"))
-
-    result = NativeGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
-
-    assert result.status == GraphRunStatus.PAUSED
-    assert result.failure_reason == "AWAITING_APPROVAL"

tests/test_task_execution_native_graph.py

@@ -1,112 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from agents.providers import DeterministicCodingProvider
-from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
-from control_plane.events.models import Event, EventType
-from control_plane.projects.models import CommitRecord, Project, ProjectPlan, Milestone, Task, TaskStatus
-from control_plane.verification.models import Review, TestRun, Verification, VerificationResult
-from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphEdgeTraversal, GraphRun, GraphRunStatus
-from graph.native_runtime import NativeGraphRuntime
-from graph.task_execution import task_execution_graph_v1
-from graph.task_nodes import TaskExecutionServices, task_execution_registry
-from model_router.router import ModelRouter
-from tests.test_m2_autonomous_loop import create_disposable_django_repo
-
-
-def create_task(repository_path: Path, goal: str, acceptance: list[str], *, max_retries: int = 2) -> Task:
-    project = Project.objects.create(name=f"Graph Project {goal[:12]}", goal=goal, repository_path=str(repository_path))
-    plan = ProjectPlan.objects.create(project=project, version=1, goal=goal)
-    milestone = Milestone.objects.create(project=project, plan=plan, key="G1", title="Graph", goal="Execution graph")
-    return Task.objects.create(
-        project=project,
-        milestone=milestone,
-        task_type="implementation",
-        status=TaskStatus.RUNNING,
-        goal=goal,
-        acceptance_criteria=acceptance,
-        max_retries=max_retries,
-    )
-
-
-def graph_run_for_task(task: Task) -> GraphRun:
-    spec = task_execution_graph_v1()
-    definition, _ = ExecutionGraphDefinition.objects.get_or_create(
-        name=spec.name,
-        defaults={"graph_type": spec.graph_type, "description": "Task execution graph"},
-    )
-    version, _ = ExecutionGraphVersion.objects.get_or_create(
-        graph=definition,
-        version=spec.version,
-        defaults={"status": ExecutionGraphVersionStatus.CHAMPION, "graph_spec": spec.to_dict()},
-    )
-    return GraphRun.objects.create(
-        execution_graph_version=version,
-        project=task.project,
-        milestone=task.milestone,
-        feature=task.feature,
-        task=task,
-        current_node=spec.entry,
-    )
-
-
-def run_graph(task: Task, *, interrupt_after: str | None = None) -> GraphRun:
-    SeedAgentsCommand().handle()
-    services = TaskExecutionServices(ModelRouter({"qwen": DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
-    runtime = NativeGraphRuntime(task_execution_registry(services))
-    return runtime.run_until_terminal_or_paused(graph_run_for_task(task), interrupt_after=interrupt_after)
-
-
-def test_native_task_execution_graph_success_matches_loop_semantics(tmp_path: Path) -> None:
-    repo = create_disposable_django_repo(tmp_path)
-    task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
-
-    graph_run = run_graph(task)
-
-    task.refresh_from_db()
-    assert graph_run.status == GraphRunStatus.COMPLETE
-    assert task.status == TaskStatus.COMPLETE
-    assert CommitRecord.objects.filter(task=task).count() == 1
-    assert TestRun.objects.get(task=task).status == "PASS"
-    assert Review.objects.get(task=task).status == "PASS"
-    assert Verification.objects.get(task=task).result == VerificationResult.PASS
-    assert task.worktree.status == "CLEANED"
-    assert GraphEdgeTraversal.objects.filter(graph_run=graph_run, source_node="judge", target_node="commit", condition="PASS").exists()
-
-
-def test_native_task_execution_graph_retry_exhaustion_matches_loop_semantics(tmp_path: Path) -> None:
-    repo = create_disposable_django_repo(tmp_path)
-    task = create_task(repo, "FORCE_BAD_IMPLEMENTATION Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"], max_retries=1)
-
-    graph_run = run_graph(task)
-
-    task.refresh_from_db()
-    assert graph_run.status == GraphRunStatus.FAILED
-    assert task.status == TaskStatus.FAILED
-    assert task.retry_count == 2
-    assert CommitRecord.objects.filter(task=task).count() == 0
-    assert Event.objects.filter(task=task, event_type=EventType.TASK_FAILED).exists()
-    assert Event.objects.filter(task=task, event_type="TASK_RETRY_EXHAUSTED").exists()
-
-
-def test_native_task_execution_graph_resume_after_coder_prevents_duplicate_commit(tmp_path: Path) -> None:
-    repo = create_disposable_django_repo(tmp_path)
-    task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
-    SeedAgentsCommand().handle()
-    graph_run = graph_run_for_task(task)
-    services = TaskExecutionServices(ModelRouter({"qwen": DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
-    runtime = NativeGraphRuntime(task_execution_registry(services))
-
-    runtime.run_until_terminal_or_paused(graph_run, interrupt_after="coder")
-    graph_run.refresh_from_db()
-    assert graph_run.current_node == "coder"
-    assert task.attempts.count() == 1
-
-    runtime.run_until_terminal_or_paused(graph_run)
-    runtime.run_until_terminal_or_paused(graph_run)
-
-    task.refresh_from_db()
-    assert task.status == TaskStatus.COMPLETE
-    assert CommitRecord.objects.filter(task=task).count() == 1
-    assert graph_run.node_runs.filter(node_id="coder").count() == 1