Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions src/db/migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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",
Expand Down
18 changes: 17 additions & 1 deletion src/local-agent-manager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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",
Expand All @@ -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",
Expand All @@ -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");
Expand Down
47 changes: 21 additions & 26 deletions src/local-agent-manager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;
Expand All @@ -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 = {
Expand All @@ -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", {
Expand All @@ -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,
Expand All @@ -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,
Expand Down
77 changes: 76 additions & 1 deletion src/local-agent-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,75 @@ 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);
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",
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);
Expand All @@ -69,6 +137,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 });
Expand Down Expand Up @@ -137,6 +206,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();
Expand Down
Loading
Loading