diff --git a/artifex/settings.py b/artifex/settings.py index 726805a..c2c563c 100644 --- a/artifex/settings.py +++ b/artifex/settings.py @@ -24,6 +24,7 @@ INSTALLED_APPS = [ "control_plane.secrets", "control_plane.knowledge", "control_plane.verification", + "graph", ] MIDDLEWARE = [ diff --git a/docs/graph_runtime_audit.md b/docs/graph_runtime_audit.md new file mode 100644 index 0000000..6db1235 --- /dev/null +++ b/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. diff --git a/graph/apps.py b/graph/apps.py new file mode 100644 index 0000000..137a8f3 --- /dev/null +++ b/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" diff --git a/graph/migrations/0001_initial.py b/graph/migrations/0001_initial.py new file mode 100644 index 0000000..d29e2a4 --- /dev/null +++ b/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")), + ] diff --git a/graph/migrations/__init__.py b/graph/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/graph/models.py b/graph/models.py new file mode 100644 index 0000000..7edf5ad --- /dev/null +++ b/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"])] diff --git a/graph/registry.py b/graph/registry.py new file mode 100644 index 0000000..aec35c8 --- /dev/null +++ b/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 diff --git a/graph/spec.py b/graph/spec.py new file mode 100644 index 0000000..87fcb19 --- /dev/null +++ b/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] diff --git a/graph/task_execution.py b/graph/task_execution.py new file mode 100644 index 0000000..52e30b4 --- /dev/null +++ b/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 diff --git a/tests/test_execution_graph_phase_a.py b/tests/test_execution_graph_phase_a.py new file mode 100644 index 0000000..f0c567e --- /dev/null +++ b/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