From 101c99fcc8b706c3d34c404a84ed207d3aa32be7 Mon Sep 17 00:00:00 2001 From: Edward Twumasi <45181928+teckedd-code2save@users.noreply.github.com> Date: Mon, 28 Sep 2026 13:42:19 +0000 Subject: [PATCH 1/2] feat: add durable agent job store --- src/core/agent-job.ts | 253 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 src/core/agent-job.ts diff --git a/src/core/agent-job.ts b/src/core/agent-job.ts new file mode 100644 index 0000000..54cdb3c --- /dev/null +++ b/src/core/agent-job.ts @@ -0,0 +1,253 @@ +import Database from 'better-sqlite3'; +import { mkdirSync } from 'node:fs'; +import { randomUUID } from 'node:crypto'; +import { resolve } from 'node:path'; + +export type AgentJobStatus = + | 'queued' + | 'running' + | 'input_required' + | 'verifying' + | 'completed' + | 'failed' + | 'cancelled'; + +export interface AgentJob { + id: string; + goal: string; + status: AgentJobStatus; + harnessProvider: string | null; + executionProvider: string | null; + capabilities: string[]; + budget: Record | null; + policy: Record | null; + createdAt: string; + updatedAt: string; +} + +export interface AgentTask { + id: string; + jobId: string; + parentTaskId: string | null; + title: string; + status: AgentJobStatus; + harnessProvider: string | null; + executionProvider: string | null; + checkpoint: Record | null; + artifacts: Array>; + error: string | null; + attempts: number; + createdAt: string; + updatedAt: string; +} + +const ALLOWED: Record = { + queued: ['running', 'cancelled'], + running: ['input_required', 'verifying', 'failed', 'cancelled'], + input_required: ['running', 'cancelled', 'failed'], + verifying: ['completed', 'running', 'failed', 'cancelled'], + completed: [], + failed: ['queued', 'running', 'cancelled'], + cancelled: [], +}; + +function assertTransition(from: AgentJobStatus, to: AgentJobStatus): void { + if (!ALLOWED[from].includes(to)) { + throw new Error(`Invalid agent job transition: ${from} -> ${to}`); + } +} + +function parseJson(value: string | null): T | null { + if (!value) return null; + return JSON.parse(value) as T; +} + +export class AgentJobStore { + private readonly db: Database.Database; + + constructor(repoPath: string) { + const dir = resolve(repoPath, '.convoy'); + mkdirSync(dir, { recursive: true }); + this.db = new Database(resolve(dir, 'state.db')); + this.db.pragma('foreign_keys = ON'); + this.db.exec(` + CREATE TABLE IF NOT EXISTS agent_jobs ( + id TEXT PRIMARY KEY, + goal TEXT NOT NULL, + status TEXT NOT NULL, + harness_provider TEXT, + execution_provider TEXT, + capabilities_json TEXT NOT NULL, + budget_json TEXT, + policy_json TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS agent_tasks ( + id TEXT PRIMARY KEY, + job_id TEXT NOT NULL REFERENCES agent_jobs(id) ON DELETE CASCADE, + parent_task_id TEXT REFERENCES agent_tasks(id) ON DELETE SET NULL, + title TEXT NOT NULL, + status TEXT NOT NULL, + harness_provider TEXT, + execution_provider TEXT, + checkpoint_json TEXT, + artifacts_json TEXT NOT NULL, + error_text TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_agent_tasks_job ON agent_tasks(job_id); + CREATE INDEX IF NOT EXISTS idx_agent_tasks_parent ON agent_tasks(parent_task_id); + `); + } + + createJob(input: { + goal: string; + harnessProvider?: string; + executionProvider?: string; + capabilities?: string[]; + budget?: Record; + policy?: Record; + }): AgentJob { + const now = new Date().toISOString(); + const id = randomUUID(); + this.db.prepare(` + INSERT INTO agent_jobs ( + id, goal, status, harness_provider, execution_provider, + capabilities_json, budget_json, policy_json, created_at, updated_at + ) VALUES (?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?) + `).run( + id, + input.goal, + input.harnessProvider ?? null, + input.executionProvider ?? null, + JSON.stringify(input.capabilities ?? []), + input.budget ? JSON.stringify(input.budget) : null, + input.policy ? JSON.stringify(input.policy) : null, + now, + now, + ); + return this.getJob(id)!; + } + + getJob(id: string): AgentJob | null { + const row = this.db.prepare('SELECT * FROM agent_jobs WHERE id = ?').get(id) as Record | undefined; + if (!row) return null; + return { + id: String(row['id']), + goal: String(row['goal']), + status: row['status'] as AgentJobStatus, + harnessProvider: (row['harness_provider'] as string | null) ?? null, + executionProvider: (row['execution_provider'] as string | null) ?? null, + capabilities: parseJson(row['capabilities_json'] as string) ?? [], + budget: parseJson>(row['budget_json'] as string | null), + policy: parseJson>(row['policy_json'] as string | null), + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + }; + } + + transitionJob(id: string, to: AgentJobStatus): AgentJob { + const current = this.getJob(id); + if (!current) throw new Error(`Agent job not found: ${id}`); + assertTransition(current.status, to); + this.db.prepare('UPDATE agent_jobs SET status = ?, updated_at = ? WHERE id = ?') + .run(to, new Date().toISOString(), id); + return this.getJob(id)!; + } + + createTask(jobId: string, input: { + title: string; + parentTaskId?: string; + harnessProvider?: string; + executionProvider?: string; + }): AgentTask { + if (!this.getJob(jobId)) throw new Error(`Agent job not found: ${jobId}`); + if (input.parentTaskId) { + const parent = this.getTask(input.parentTaskId); + if (!parent || parent.jobId !== jobId) { + throw new Error('Parent task must belong to the same job'); + } + } + const id = randomUUID(); + const now = new Date().toISOString(); + this.db.prepare(` + INSERT INTO agent_tasks ( + id, job_id, parent_task_id, title, status, harness_provider, + execution_provider, checkpoint_json, artifacts_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, 'queued', ?, ?, NULL, '[]', ?, ?) + `).run( + id, + jobId, + input.parentTaskId ?? null, + input.title, + input.harnessProvider ?? null, + input.executionProvider ?? null, + now, + now, + ); + return this.getTask(id)!; + } + + getTask(id: string): AgentTask | null { + const row = this.db.prepare('SELECT * FROM agent_tasks WHERE id = ?').get(id) as Record | undefined; + if (!row) return null; + return { + id: String(row['id']), + jobId: String(row['job_id']), + parentTaskId: (row['parent_task_id'] as string | null) ?? null, + title: String(row['title']), + status: row['status'] as AgentJobStatus, + harnessProvider: (row['harness_provider'] as string | null) ?? null, + executionProvider: (row['execution_provider'] as string | null) ?? null, + checkpoint: parseJson>(row['checkpoint_json'] as string | null), + artifacts: parseJson>>(row['artifacts_json'] as string) ?? [], + error: (row['error_text'] as string | null) ?? null, + attempts: Number(row['attempts']), + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + }; + } + + listTasks(jobId: string): AgentTask[] { + const rows = this.db.prepare('SELECT id FROM agent_tasks WHERE job_id = ? ORDER BY created_at ASC').all(jobId) as Array<{ id: string }>; + return rows.map(row => this.getTask(row.id)!).filter(Boolean); + } + + transitionTask(id: string, to: AgentJobStatus, error?: string): AgentTask { + const current = this.getTask(id); + if (!current) throw new Error(`Agent task not found: ${id}`); + assertTransition(current.status, to); + const attempts = to === 'running' ? current.attempts + 1 : current.attempts; + this.db.prepare(` + UPDATE agent_tasks + SET status = ?, error_text = ?, attempts = ?, updated_at = ? + WHERE id = ? + `).run(to, error ?? null, attempts, new Date().toISOString(), id); + return this.getTask(id)!; + } + + checkpointTask(id: string, checkpoint: Record): AgentTask { + if (!this.getTask(id)) throw new Error(`Agent task not found: ${id}`); + this.db.prepare('UPDATE agent_tasks SET checkpoint_json = ?, updated_at = ? WHERE id = ?') + .run(JSON.stringify(checkpoint), new Date().toISOString(), id); + return this.getTask(id)!; + } + + addArtifact(id: string, artifact: Record): AgentTask { + const current = this.getTask(id); + if (!current) throw new Error(`Agent task not found: ${id}`); + const artifacts = [...current.artifacts, artifact]; + this.db.prepare('UPDATE agent_tasks SET artifacts_json = ?, updated_at = ? WHERE id = ?') + .run(JSON.stringify(artifacts), new Date().toISOString(), id); + return this.getTask(id)!; + } + + close(): void { + this.db.close(); + } +} From 0e4de3100c5af73ecc3af5d3ee1f7e9d12a387d3 Mon Sep 17 00:00:00 2001 From: Edward Twumasi <45181928+teckedd-code2save@users.noreply.github.com> Date: Mon, 28 Sep 2026 13:42:21 +0000 Subject: [PATCH 2/2] test: cover durable agent job state --- src/core/agent-job.test.ts | 74 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 src/core/agent-job.test.ts diff --git a/src/core/agent-job.test.ts b/src/core/agent-job.test.ts new file mode 100644 index 0000000..68f48c4 --- /dev/null +++ b/src/core/agent-job.test.ts @@ -0,0 +1,74 @@ +import test, { afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; + +import { AgentJobStore } from './agent-job.js'; + +const tempDirs: string[] = []; +function tmp(): string { + const d = mkdtempSync(join(tmpdir(), 'convoy-agent-job-')); + tempDirs.push(d); + return d; +} +afterEach(() => { + while (tempDirs.length) { + const d = tempDirs.pop(); + if (d) rmSync(d, { recursive: true, force: true }); + } +}); + +test('persists jobs, child tasks, checkpoints and evidence across store restarts', () => { + const repo = tmp(); + const store = new AgentJobStore(repo); + const job = store.createJob({ + goal: 'ship the API safely', + capabilities: ['repo.inspect', 'deploy.verify'], + }); + store.transitionJob(job.id, 'running'); + + const parent = store.createTask(job.id, { title: 'inspect repo' }); + store.transitionTask(parent.id, 'running'); + store.checkpointTask(parent.id, { scanned: true }); + store.addArtifact(parent.id, { kind: 'repo-scan', services: 3 }); + + const child = store.createTask(job.id, { + title: 'verify deployment', + parentTaskId: parent.id, + }); + store.close(); + + const reopened = new AgentJobStore(repo); + assert.equal(reopened.getJob(job.id)?.status, 'running'); + assert.deepEqual(reopened.getTask(parent.id)?.checkpoint, { scanned: true }); + assert.equal(reopened.getTask(parent.id)?.artifacts[0]?.['kind'], 'repo-scan'); + assert.equal(reopened.getTask(child.id)?.parentTaskId, parent.id); + reopened.close(); +}); + +test('rejects invalid transitions and allows retry from failed', () => { + const repo = tmp(); + const store = new AgentJobStore(repo); + const job = store.createJob({ goal: 'diagnose production' }); + assert.throws(() => store.transitionJob(job.id, 'completed'), /Invalid agent job transition/); + + store.transitionJob(job.id, 'running'); + store.transitionJob(job.id, 'failed'); + const retried = store.transitionJob(job.id, 'running'); + assert.equal(retried.status, 'running'); + store.close(); +}); + +test('parent task must belong to the same job', () => { + const repo = tmp(); + const store = new AgentJobStore(repo); + const a = store.createJob({ goal: 'a' }); + const b = store.createJob({ goal: 'b' }); + const parent = store.createTask(a.id, { title: 'parent' }); + assert.throws( + () => store.createTask(b.id, { title: 'child', parentTaskId: parent.id }), + /same job/, + ); + store.close(); +});