Add execution graph domain model

artifex/settings.py

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

docs/graph_runtime_audit.md

@@ -0,0 +1,524 @@
+# 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

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

graph/migrations/0001_initial.py

@@ -0,0 +1,100 @@
+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/models.py

@@ -0,0 +1,109 @@
+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)
+    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"], name="unique_graph_node_run")]
+        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"])]

graph/registry.py

@@ -0,0 +1,40 @@
+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/spec.py

@@ -0,0 +1,91 @@
+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

@@ -0,0 +1,54 @@
+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

tests/test_execution_graph_phase_a.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import pytest
+
+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