From 2a485d6c91637b215bff8b2128101a2c08264bea Mon Sep 17 00:00:00 2001 From: Waishnav <86405648+Waishnav@users.noreply.github.com> Date: Mon, 31 Aug 2026 03:51:30 +0530 Subject: [PATCH 1/2] feat(agents): persist internal turn history --- src/db/migrations.ts | 29 +++++ src/local-agent-manager.test.ts | 18 ++- src/local-agent-manager.ts | 47 ++++--- src/local-agent-store.test.ts | 68 +++++++++- src/local-agent-store.ts | 217 ++++++++++++++++++++++++++++++-- src/oauth-store.test.ts | 1 + 6 files changed, 343 insertions(+), 37 deletions(-) diff --git a/src/db/migrations.ts b/src/db/migrations.ts index df192caa0..cf19f8efa 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -37,6 +37,11 @@ const migrations: Migration[] = [ name: "local-agent-effort-rename", up: migrateLocalAgentEffortRename, }, + { + version: 7, + name: "local-agent-turns", + up: migrateLocalAgentTurns, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -235,6 +240,30 @@ function migrateLocalAgentEffortRename(sqlite: Database.Database): void { sqlite.exec("alter table local_agent_sessions rename column thinking to effort"); } +function migrateLocalAgentTurns(sqlite: Database.Database): void { + sqlite.exec(` + create table if not exists local_agent_turns ( + id integer primary key autoincrement, + agent_id text not null, + prompt text not null, + status text not null, + response text, + error text, + error_code text, + error_retryable text, + created_at text not null, + completed_at text, + foreign key (agent_id) references local_agent_sessions(id) on delete cascade + ); + + create index if not exists local_agent_turns_agent_id_idx + on local_agent_turns(agent_id, id desc); + + create index if not exists local_agent_turns_status_idx + on local_agent_turns(status); + `); +} + function addColumnIfMissing( sqlite: Database.Database, table: "workspace_sessions" | "local_agent_sessions", diff --git a/src/local-agent-manager.test.ts b/src/local-agent-manager.test.ts index 4ca5ed28d..803866fb9 100644 --- a/src/local-agent-manager.test.ts +++ b/src/local-agent-manager.test.ts @@ -121,7 +121,8 @@ const stale = store.create({ profileName: "reviewer", provider: "codex", }); -store.update(stale.id, { status: "running", latestResponse: "previous response" }); +const staleTurn = store.beginTurn(stale.id, { prompt: "interrupted turn" }); +store.update(stale.id, { latestResponse: "previous response" }); const manager = new LocalAgentManager({ store, @@ -222,6 +223,8 @@ assert.equal(getRecord(stale.id).latestResponse, "previous response"); assert.equal(getRecord(stale.id).error, "DevSpace restarted while this agent turn was running."); assert.equal(getRecord(stale.id).errorCode, "DAEMON_UNAVAILABLE"); assert.equal(getRecord(stale.id).errorRetryable, true); +assert.equal(store.getTurnById(staleTurn.turn.id)?.status, "failed"); +assert.equal(store.getTurnById(staleTurn.turn.id)?.errorCode, "DAEMON_UNAVAILABLE"); const first = unwrap(await manager.start({ target: "reviewer", @@ -244,6 +247,10 @@ runtimes.get(first.id)!.release(); await waitFor(() => getRecord(first.id).status === "idle"); assert.equal(getRecord(first.id).providerSessionId, "thread_test"); assert.match(getRecord(first.id).latestResponse ?? "", /Task:\nhold/); +assert.deepEqual( + store.listTurns(first.id).map((turn) => ({ prompt: turn.prompt, status: turn.status })), + [{ prompt: "hold", status: "completed" }], +); const continued = unwrap(await manager.continue(first.id, "continue", { model: "gpt-run", @@ -253,6 +260,13 @@ assert.equal(continued.status, "running"); await waitFor(() => getRecord(first.id).status === "idle"); assert.equal(getRecord(first.id).model, "gpt-run"); assert.equal(getRecord(first.id).effort, "high"); +assert.deepEqual( + store.listTurns(first.id).map((turn) => ({ prompt: turn.prompt, status: turn.status })), + [ + { prompt: "hold", status: "completed" }, + { prompt: "continue", status: "completed" }, + ], +); const second = unwrap(await manager.start({ target: "reviewer", @@ -274,6 +288,8 @@ await waitFor(() => getRecord(failed.id).status === "error"); assert.equal(getRecord(failed.id).error, "provider failed"); assert.equal(getRecord(failed.id).errorCode, "PROVIDER_EXECUTION_ERROR"); assert.equal(getRecord(failed.id).errorRetryable, false); +assert.equal(store.getLatestTurn(failed.id)?.status, "failed"); +assert.equal(store.getLatestTurn(failed.id)?.error, "provider failed"); const recovered = unwrap(await manager.continue(failed.id, "recovered", {}, scope)); assert.equal(recovered.status, "running", "provider Err releases active-turn ownership"); await waitFor(() => getRecord(failed.id).status === "idle"); diff --git a/src/local-agent-manager.ts b/src/local-agent-manager.ts index dbd80d86b..ef8a3c720 100644 --- a/src/local-agent-manager.ts +++ b/src/local-agent-manager.ts @@ -246,28 +246,25 @@ export class LocalAgentManager { })); } - const updated = this.store.updateResult(record.id, { - status: "running", + const begun = this.store.beginTurnResult(record.id, { + prompt, model: overrides.model ?? record.model, effort: overrides.effort ?? record.effort, - latestResponse: undefined, - error: undefined, - errorCode: undefined, - errorRetryable: undefined, }); - if (updated.isErr()) return updated; + if (begun.isErr()) return begun; // Defer invocation until after the tracking entry is visible. This keeps // cleanup correct even if runTurn later gains a synchronous completion path. const turn = Promise.resolve().then(() => ( - this.runTurn(updated.value, prompt, overrides, workspaceId) + this.runTurn(begun.value.agent, begun.value.turn.id, prompt, overrides, workspaceId) )); this.activeTurns.set(record.id, turn); void turn.catch(() => undefined); - return updated; + return Result.ok(begun.value.agent); } private async runTurn( record: LocalAgentRecord, + turnId: number, prompt: string, overrides: RunOverrides, workspaceId?: string, @@ -281,7 +278,7 @@ export class LocalAgentManager { try { const authorized = this.authorizeWorkspace(record.workspaceRoot, workspaceId, "run"); if (authorized.isErr()) { - this.persistRunError(record, authorized.error, startedAt); + this.persistRunError(record, turnId, authorized.error, startedAt); return; } const workspaceRoot = authorized.value; @@ -290,22 +287,22 @@ export class LocalAgentManager { : { ...record, workspaceRoot }; const profiles = await this.loadProfilesResult(workspaceRoot, record.profileName); if (profiles.isErr()) { - this.persistRunError(record, profiles.error, startedAt); + this.persistRunError(record, turnId, profiles.error, startedAt); return; } const profile = this.profileForRecordResult(record, profiles.value); if (profile.isErr()) { - this.persistRunError(record, profile.error, startedAt); + this.persistRunError(record, turnId, profile.error, startedAt); return; } const input = this.buildRunInputResult(authorizedRecord, profile.value, prompt, overrides); if (input.isErr()) { - this.persistRunError(record, input.error, startedAt); + this.persistRunError(record, turnId, input.error, startedAt); return; } const driver = this.driverResult(record.provider, "run", record.id); if (driver.isErr()) { - this.persistRunError(record, driver.error, startedAt); + this.persistRunError(record, turnId, driver.error, startedAt); return; } const context: LocalAgentRuntimeContext = { @@ -329,20 +326,17 @@ export class LocalAgentManager { }; const result = await this.pool.run(driver.value, context, input.value, callbacks); if (result.isErr()) { - this.persistRunError(record, result.error, startedAt); + this.persistRunError(record, turnId, result.error, startedAt); return; } const runResult = result.value; const current = this.store.getByIdResult(record.id); if (current.isErr()) throw current.error; if (!current.value) return; - const updated = this.store.updateResult(record.id, { + const updated = this.store.finishTurnResult(record.id, turnId, { providerSessionId: runResult.providerSessionId ?? current.value.providerSessionId, - status: "idle", - latestResponse: runResult.finalResponse, - error: undefined, - errorCode: undefined, - errorRetryable: undefined, + status: "completed", + response: runResult.finalResponse, }); if (updated.isErr()) throw updated.error; this.log("info", "agent_run_completed", { @@ -353,11 +347,11 @@ export class LocalAgentManager { }); } catch (error) { if (isLocalAgentError(error)) { - this.persistRunError(record, error, startedAt); + this.persistRunError(record, turnId, error, startedAt); return; } - const persisted = this.store.updateResult(record.id, { - status: "error", + const persisted = this.store.finishTurnResult(record.id, turnId, { + status: "failed", error: "Unexpected internal subagent failure.", errorCode: "AGENT_INTERNAL_ERROR", errorRetryable: false, @@ -379,11 +373,12 @@ export class LocalAgentManager { private persistRunError( record: LocalAgentRecord, + turnId: number, error: LocalAgentError, startedAt: number, ): void { - const persisted = this.store.updateResult(record.id, { - status: "error", + const persisted = this.store.finishTurnResult(record.id, turnId, { + status: "failed", error: error.message, errorCode: error.code, errorRetryable: error.retryable, diff --git a/src/local-agent-store.test.ts b/src/local-agent-store.test.ts index 829940f92..7f4d9440f 100644 --- a/src/local-agent-store.test.ts +++ b/src/local-agent-store.test.ts @@ -54,7 +54,66 @@ try { assert.deepEqual(store.list({ workspaceId: "ws_1" }).map((agent) => agent.id), [created.id]); assert.deepEqual(store.list({ workspaceId: "ws_other" }), []); assert.deepEqual(store.list({ workspaceId: "ws_1", workspaceRoot: join(root, "other") }), []); -assert.deepEqual(store.list({ workspaceRoot: join(root, "other") }), []); + assert.deepEqual(store.list({ workspaceRoot: join(root, "other") }), []); + + const begun = store.beginTurn(created.id, { + prompt: "Review the current changes.", + model: updated.model, + effort: updated.effort, + }); + assert.equal(begun.agent.status, "running"); + assert.equal(begun.turn.agentId, created.id); + assert.equal(begun.turn.prompt, "Review the current changes."); + assert.equal(begun.turn.status, "running"); + assert.equal(begun.turn.completedAt, undefined); + + const completed = store.finishTurn(created.id, begun.turn.id, { + status: "completed", + response: "No issues found.", + providerSessionId: "thread_456", + }); + assert.equal(completed.status, "idle"); + assert.equal(completed.latestResponse, "No issues found."); + assert.equal(completed.providerSessionId, "thread_456"); + const completedTurn = store.getLatestTurn(created.id); + assert.equal(completedTurn?.id, begun.turn.id); + assert.equal(completedTurn?.status, "completed"); + assert.equal(completedTurn?.response, "No issues found."); + assert.ok(completedTurn?.completedAt); + + const failing = store.beginTurn(created.id, { + prompt: "Retry the review.", + model: completed.model, + effort: completed.effort, + }); + store.finishTurn(created.id, failing.turn.id, { + status: "failed", + error: "Provider disconnected.", + errorCode: "PROVIDER_EXECUTION_ERROR", + errorRetryable: true, + }); + assert.deepEqual( + store.listTurns(created.id).map((turn) => ({ + prompt: turn.prompt, + status: turn.status, + response: turn.response, + errorCode: turn.errorCode, + })), + [ + { + prompt: "Review the current changes.", + status: "completed", + response: "No issues found.", + errorCode: undefined, + }, + { + prompt: "Retry the review.", + status: "failed", + response: undefined, + errorCode: "PROVIDER_EXECUTION_ERROR", + }, + ], + ); const otherStore = new LocalAgentStore(root); stores.push(otherStore); @@ -69,6 +128,7 @@ assert.deepEqual(store.list({ workspaceRoot: join(root, "other") }), []); store.list({ workspaceId: "ws_1" }).map((agent) => agent.id).sort(), [created.id, createdFromOtherStore.id].sort(), ); + assert.equal(otherStore.listTurns(created.id).length, 2); const legacyStateDir = join(root, "legacy-state"); mkdirSync(legacyStateDir, { recursive: true }); @@ -137,6 +197,12 @@ assert.deepEqual(store.list({ workspaceRoot: join(root, "other") }), []); assert.equal(reloadedRecord?.error, "old error"); assert.equal(reloadedRecord?.errorCode, "DAEMON_TIMEOUT"); assert.equal(reloadedRecord?.errorRetryable, true); + const legacyTurn = upgradedStore.beginTurn("agt_legacy", { + prompt: "Continue after upgrade.", + model: reloadedRecord?.model, + effort: reloadedRecord?.effort, + }); + assert.equal(legacyTurn.turn.status, "running"); } finally { for (const store of stores) { store.close(); diff --git a/src/local-agent-store.ts b/src/local-agent-store.ts index 74bf875d5..f3fa93c3f 100644 --- a/src/local-agent-store.ts +++ b/src/local-agent-store.ts @@ -5,6 +5,7 @@ import { openDatabase, type DatabaseHandle } from "./db/client.js"; import { AgentStoreError, isProgrammerDefect } from "./local-agent-errors.js"; export type LocalAgentStatus = "starting" | "running" | "idle" | "error" | "stopped"; +export type LocalAgentTurnStatus = "running" | "completed" | "failed" | "stopped"; export interface LocalAgentRecord { id: string; @@ -33,6 +34,35 @@ export interface CreateLocalAgentRecordInput { effort?: string; } +export interface LocalAgentTurnRecord { + id: number; + agentId: string; + prompt: string; + status: LocalAgentTurnStatus; + response?: string; + error?: string; + errorCode?: string; + errorRetryable?: boolean; + createdAt: string; + completedAt?: string; +} + +export interface BeginLocalAgentTurnInput { + prompt: string; + model?: string; + effort?: string; +} + +export type FinishLocalAgentTurnInput = + | { status: "completed"; response?: string; providerSessionId?: string } + | { status: "failed"; error: string; errorCode: string; errorRetryable: boolean } + | { status: "stopped"; error?: string; errorCode?: string; errorRetryable?: boolean }; + +export interface BegunLocalAgentTurn { + agent: LocalAgentRecord; + turn: LocalAgentTurnRecord; +} + export interface LocalAgentWorkspaceScope { workspaceId?: string; workspaceRoot: string; @@ -61,6 +91,19 @@ interface LocalAgentRow { updated_at: string; } +interface LocalAgentTurnRow { + id: number; + agent_id: string; + prompt: string; + status: string; + response: string | null; + error: string | null; + error_code: string | null; + error_retryable: string | null; + created_at: string; + completed_at: string | null; +} + export class LocalAgentStore { private readonly database: DatabaseHandle; @@ -235,16 +278,150 @@ export class LocalAgentStore { return storeResult("update", () => this.update(id, patch)); } + beginTurn(agentId: string, input: BeginLocalAgentTurnInput): BegunLocalAgentTurn { + return this.database.sqlite.transaction(() => { + const agent = this.update(agentId, { + status: "running", + model: input.model, + effort: input.effort, + latestResponse: undefined, + error: undefined, + errorCode: undefined, + errorRetryable: undefined, + }); + const result = this.database.sqlite + .prepare( + `insert into local_agent_turns ( + agent_id, + prompt, + status, + created_at + ) values (?, ?, 'running', ?)`, + ) + .run(agentId, input.prompt, agent.updatedAt); + const turn = this.getTurnById(Number(result.lastInsertRowid)); + if (!turn) throw new Error(`Unable to load the new turn for subagent ${agentId}.`); + return { agent, turn }; + }).immediate(); + } + + beginTurnResult( + agentId: string, + input: BeginLocalAgentTurnInput, + ): BetterResult { + return storeResult("begin_turn", () => this.beginTurn(agentId, input)); + } + + finishTurn( + agentId: string, + turnId: number, + completion: FinishLocalAgentTurnInput, + ): LocalAgentRecord { + return this.database.sqlite.transaction(() => { + const turn = this.getTurnById(turnId); + if (!turn || turn.agentId !== agentId) { + throw new Error(`Unknown turn ${turnId} for subagent ${agentId}.`); + } + if (turn.status !== "running") { + throw new Error(`Turn ${turnId} for subagent ${agentId} is already ${turn.status}.`); + } + const currentAgent = this.getById(agentId); + if (!currentAgent) throw new Error(`Unknown subagent id: ${agentId}`); + + const completedAt = new Date().toISOString(); + this.database.sqlite + .prepare( + `update local_agent_turns set + status = ?, + response = ?, + error = ?, + error_code = ?, + error_retryable = ?, + completed_at = ? + where id = ? and agent_id = ?`, + ) + .run( + completion.status, + completion.status === "completed" ? completion.response ?? null : null, + completion.status === "completed" ? null : completion.error ?? null, + completion.status === "completed" ? null : completion.errorCode ?? null, + completion.status === "completed" || completion.errorRetryable === undefined + ? null + : String(completion.errorRetryable), + completedAt, + turnId, + agentId, + ); + + if (completion.status === "completed") { + return this.update(agentId, { + providerSessionId: completion.providerSessionId ?? currentAgent.providerSessionId, + status: "idle", + latestResponse: completion.response, + error: undefined, + errorCode: undefined, + errorRetryable: undefined, + }); + } + return this.update(agentId, { + status: completion.status === "failed" ? "error" : "stopped", + latestResponse: undefined, + error: completion.error, + errorCode: completion.errorCode, + errorRetryable: completion.errorRetryable, + }); + }).immediate(); + } + + finishTurnResult( + agentId: string, + turnId: number, + completion: FinishLocalAgentTurnInput, + ): BetterResult { + return storeResult("finish_turn", () => this.finishTurn(agentId, turnId, completion)); + } + + getTurnById(turnId: number): LocalAgentTurnRecord | undefined { + const row = this.database.sqlite + .prepare("select * from local_agent_turns where id = ? limit 1") + .get(turnId) as LocalAgentTurnRow | undefined; + return row ? rowToLocalAgentTurnRecord(row) : undefined; + } + + getLatestTurn(agentId: string): LocalAgentTurnRecord | undefined { + const row = this.database.sqlite + .prepare("select * from local_agent_turns where agent_id = ? order by id desc limit 1") + .get(agentId) as LocalAgentTurnRow | undefined; + return row ? rowToLocalAgentTurnRecord(row) : undefined; + } + + listTurns(agentId: string): LocalAgentTurnRecord[] { + const rows = this.database.sqlite + .prepare("select * from local_agent_turns where agent_id = ? order by id asc") + .all(agentId) as LocalAgentTurnRow[]; + return rows.map(rowToLocalAgentTurnRecord); + } + reconcileActiveRuns(message = "DevSpace restarted while this agent turn was running."): number { - const now = new Date().toISOString(); - const result = this.database.sqlite - .prepare( - `update local_agent_sessions - set status = 'error', error = ?, error_code = 'DAEMON_UNAVAILABLE', error_retryable = 'true', updated_at = ? - where status in ('starting', 'running')`, - ) - .run(message, now); - return Number(result.changes); + return this.database.sqlite.transaction(() => { + const now = new Date().toISOString(); + this.database.sqlite + .prepare( + `update local_agent_turns + set status = 'failed', error = ?, error_code = 'DAEMON_UNAVAILABLE', + error_retryable = 'true', completed_at = ? + where status = 'running'`, + ) + .run(message, now); + const result = this.database.sqlite + .prepare( + `update local_agent_sessions + set status = 'error', error = ?, error_code = 'DAEMON_UNAVAILABLE', error_retryable = 'true', updated_at = ? + where status in ('starting', 'running')`, + ) + .run(message, now); + return Number(result.changes); + }).immediate(); } reconcileActiveRunsResult( @@ -283,6 +460,28 @@ function rowToLocalAgentRecord(row: LocalAgentRow): LocalAgentRecord { }; } +function rowToLocalAgentTurnRecord(row: LocalAgentTurnRow): LocalAgentTurnRecord { + return { + id: row.id, + agentId: row.agent_id, + prompt: row.prompt, + status: readTurnStatus(row.status), + response: row.response ?? undefined, + error: row.error ?? undefined, + errorCode: row.error_code ?? undefined, + errorRetryable: readOptionalBoolean(row.error_retryable), + createdAt: row.created_at, + completedAt: row.completed_at ?? undefined, + }; +} + +function readTurnStatus(status: string): LocalAgentTurnStatus { + if (status === "running" || status === "completed" || status === "failed" || status === "stopped") { + return status; + } + throw new Error(`Invalid stored local agent turn status: ${status}`); +} + function readOptionalBoolean(value: string | null): boolean | undefined { if (value === "true") return true; if (value === "false") return false; diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index 225f9fdf5..f535234a3 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -47,6 +47,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 4, name: "workspace-conversation-bindings" }, { version: 5, name: "local-agent-structured-errors" }, { version: 6, name: "local-agent-effort-rename" }, + { version: 7, name: "local-agent-turns" }, ]); } finally { database.close(); From 50ef1f38e20d49a78380e953f133b8034494d0a9 Mon Sep 17 00:00:00 2001 From: Waishnav <86405648+Waishnav@users.noreply.github.com> Date: Mon, 31 Aug 2026 19:01:14 +0530 Subject: [PATCH 2/2] fix(agents): prevent overlapping persisted turns --- src/local-agent-store.test.ts | 9 +++++++++ src/local-agent-store.ts | 5 +++++ 2 files changed, 14 insertions(+) diff --git a/src/local-agent-store.test.ts b/src/local-agent-store.test.ts index 7f4d9440f..e29bd319b 100644 --- a/src/local-agent-store.test.ts +++ b/src/local-agent-store.test.ts @@ -66,6 +66,15 @@ assert.deepEqual(store.list({ workspaceId: "ws_1", workspaceRoot: join(root, "ot assert.equal(begun.turn.prompt, "Review the current changes."); assert.equal(begun.turn.status, "running"); assert.equal(begun.turn.completedAt, undefined); + assert.throws( + () => store.beginTurn(created.id, { + prompt: "Start overlapping work.", + model: begun.agent.model, + effort: begun.agent.effort, + }), + /already has a running turn/, + ); + assert.equal(store.listTurns(created.id).length, 1); const completed = store.finishTurn(created.id, begun.turn.id, { status: "completed", diff --git a/src/local-agent-store.ts b/src/local-agent-store.ts index f3fa93c3f..bdb014a66 100644 --- a/src/local-agent-store.ts +++ b/src/local-agent-store.ts @@ -280,6 +280,11 @@ export class LocalAgentStore { beginTurn(agentId: string, input: BeginLocalAgentTurnInput): BegunLocalAgentTurn { return this.database.sqlite.transaction(() => { + const current = this.getById(agentId); + if (!current) throw new Error(`Unknown subagent id: ${agentId}`); + if (current.status === "running") { + throw new Error(`Subagent ${agentId} already has a running turn.`); + } const agent = this.update(agentId, { status: "running", model: input.model,