Q54: later sessions see what earlier ones already did and where they failed
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
ec2698575d
commit
e809cfefea
8 files changed
+146
-2
No files matched your search
@@ -0,0 +1,16 @@
|
||||
export const runActionsSchema = `
|
||||
create table if not exists run_actions (
|
||||
id text primary key,
|
||||
org_id text not null references organizations(id) on delete cascade,
|
||||
run_id text not null references task_runs(id) on delete cascade,
|
||||
task_id text not null references tasks(id) on delete cascade,
|
||||
tool_name text not null,
|
||||
arguments_excerpt text not null,
|
||||
result_excerpt text not null,
|
||||
created_at text not null
|
||||
);
|
||||
create index if not exists idx_run_actions_task on run_actions(task_id);
|
||||
|
||||
insert into schema_migrations(version, name, applied_at)
|
||||
values (70, 'run_actions', datetime('now'));
|
||||
`;
|
||||
@@ -68,6 +68,7 @@ import { taskWritingModeSchema } from "./migrations/066-task-writing-mode";
|
||||
import { taskRunBudgetSchema } from "./migrations/067-task-run-budget";
|
||||
import { taskReviewLinkSchema } from "./migrations/068-task-review-link";
|
||||
import { goalStallSchema } from "./migrations/069-goal-stall";
|
||||
import { runActionsSchema } from "./migrations/070-run-actions";
|
||||
|
||||
interface Migration {
|
||||
version: number;
|
||||
@@ -145,6 +146,7 @@ const MIGRATIONS: Migration[] = [
|
||||
{ version: 67, name: "task_run_budget", sql: taskRunBudgetSchema },
|
||||
{ version: 68, name: "task_review_link", sql: taskReviewLinkSchema },
|
||||
{ version: 69, name: "goal_stall", sql: goalStallSchema },
|
||||
{ version: 70, name: "run_actions", sql: runActionsSchema },
|
||||
];
|
||||
|
||||
const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version;
|
||||
|
||||
@@ -4,7 +4,7 @@ import { completeGoalTool } from "./agent-goal-tools";
|
||||
import { executeWebTool, isWebTool, WEB_TOOLS } from "./agent-web-tools";
|
||||
import { executeWorkspaceTool, isWorkspaceTool, WORKSPACE_TOOLS } from "./agent-workspace-tools";
|
||||
import { canonicalMentionHandle } from "../../shared/mention-handles";
|
||||
import type { AgentTurnMessage, ToolDefinition } from "./provider-tool-turn";
|
||||
import type { AgentTurnMessage, ToolCall, ToolDefinition } from "./provider-tool-turn";
|
||||
import { isComposioTool, executeComposioTool } from "./composio/composio-tool-bridge";
|
||||
import { capToolOutput, compactToolHistory } from "./run-context-compaction";
|
||||
import { fitContextBudget } from "./context-budget";
|
||||
@@ -23,6 +23,8 @@ export interface AgentRunLoopInput {
|
||||
pollNewMessages?: () => string | null;
|
||||
/** Called after each model turn the provider answered. */
|
||||
onModelTurn?: () => void;
|
||||
/** Called with every executed tool call and its full output. */
|
||||
onToolCompleted?: (call: ToolCall, output: string) => void;
|
||||
/** Input tokens the provider's context window leaves after the output reserve. */
|
||||
inputBudgetTokens: number;
|
||||
}
|
||||
@@ -119,6 +121,7 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
|
||||
announceTaskAssignments: shouldAnnounceTaskAssignment(call, turn.toolCalls),
|
||||
});
|
||||
}
|
||||
input.onToolCompleted?.(call, output);
|
||||
results.push({ toolCallId: call.id, toolName: call.name, output: capToolOutput(call.name, output) });
|
||||
}
|
||||
const followUp = input.pollNewMessages?.();
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import { createId } from "../../shared/ids";
|
||||
import type { ToolCall } from "./provider-tool-turn";
|
||||
|
||||
/** Tools that only look; everything else changes the company or the world and is journaled. */
|
||||
const READ_ONLY_TOOLS = new Set([
|
||||
"company_status", "fetch_url", "list_connectors", "list_documents", "list_files", "list_skills",
|
||||
"read_attachment", "read_document", "read_file", "read_messages", "search_knowledge", "web_search",
|
||||
]);
|
||||
const JOURNAL_LINES = 40;
|
||||
const FIELD_CHARS = 160;
|
||||
|
||||
/** Records a call that tried to change something, with its result, so later sessions build on it instead of repeating it. */
|
||||
export function recordRunAction(db: DatabaseSync, input: { orgId: string; runId: string; taskId: string; call: ToolCall; output: string }): void {
|
||||
if (READ_ONLY_TOOLS.has(input.call.name)) return;
|
||||
db.prepare(`
|
||||
insert into run_actions(id, org_id, run_id, task_id, tool_name, arguments_excerpt, result_excerpt, created_at)
|
||||
values (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`).run(createId("rac"), input.orgId, input.runId, input.taskId, input.call.name,
|
||||
clip(JSON.stringify(input.call.arguments)), clip(input.output.split("\n")[0] ?? ""), new Date().toISOString());
|
||||
}
|
||||
|
||||
/** What earlier sessions (including an interrupted attempt of this run) already did on the task, and where the last one failed. */
|
||||
export function runJournalLines(db: DatabaseSync, taskId: string): string[] {
|
||||
const actions = db.prepare(`
|
||||
select tool_name, arguments_excerpt, result_excerpt from (
|
||||
select tool_name, arguments_excerpt, result_excerpt, rowid as seq from run_actions where task_id = ? order by rowid desc limit ?
|
||||
) order by seq asc
|
||||
`).all(taskId, JOURNAL_LINES) as Array<{ tool_name: string; arguments_excerpt: string; result_excerpt: string }>;
|
||||
const lastFailure = db.prepare(`
|
||||
select error_message from task_runs where task_id = ? and state in ('completed', 'failed') order by rowid desc limit 1
|
||||
`).get(taskId) as { error_message: string | null } | undefined;
|
||||
const lines: string[] = [];
|
||||
if (actions.length > 0) {
|
||||
lines.push("", "Earlier sessions on this task already made these calls; each shows its result. Do not redo the ones that succeeded; retry or fix only what failed:");
|
||||
for (const action of actions) lines.push(`- ${action.tool_name}(${action.arguments_excerpt}) -> ${action.result_excerpt}`);
|
||||
}
|
||||
if (lastFailure?.error_message) {
|
||||
lines.push("", `The last session on this task failed with: ${lastFailure.error_message}. Pick up from there instead of starting over.`);
|
||||
}
|
||||
return lines;
|
||||
}
|
||||
|
||||
function clip(text: string): string {
|
||||
return text.length > FIELD_CHARS ? `${text.slice(0, FIELD_CHARS)}...` : text;
|
||||
}
|
||||
@@ -12,6 +12,7 @@ import { SkillService } from "./skill-service";
|
||||
import { runAgentLoop, runTools } from "./agent-run-loop";
|
||||
import type { RunModel } from "./skill-runtime";
|
||||
import { combineMidRunPolls, createGoalChangePoller } from "./goal-change-poller";
|
||||
import { recordRunAction, runJournalLines } from "./run-journal";
|
||||
import type { AgentToolContext } from "./agent-tools";
|
||||
import {
|
||||
buildAgentInstructions,
|
||||
@@ -222,6 +223,7 @@ export class TaskExecutionService {
|
||||
handoffLines: renderTaskHandoff(this.db, orgId, task.handoff_json),
|
||||
knowledgeLines: relevantKnowledgeLines(this.db, { orgId, agentId: agent.id, projectId: task.project_id, goalId: task.goal_id, taskId: task.id }, `${task.title}\n${task.description}`),
|
||||
priorCompletedRuns: this.completedRunCount(task.id, runId),
|
||||
journalLines: runJournalLines(this.db, task.id),
|
||||
subtasks: (this.db.prepare("select title, status = 'done' as done from tasks where parent_task_id = ? order by created_at").all(task.id) as Array<{ title: string; done: number }>)
|
||||
.map((row) => ({ title: row.title, done: row.done === 1 })),
|
||||
availableToolNames: runTools(externalTools).map((tool) => tool.name),
|
||||
@@ -229,6 +231,7 @@ export class TaskExecutionService {
|
||||
toolContext,
|
||||
externalTools,
|
||||
onModelTurn: contextRecorder.commit,
|
||||
onToolCompleted: (call, output) => recordRunAction(this.db, { orgId, runId, taskId: task.id, call, output }),
|
||||
inputBudgetTokens: inputBudgetTokens(provider.context_window_tokens),
|
||||
pollNewMessages: combineMidRunPolls(
|
||||
goalChanges,
|
||||
|
||||
@@ -201,6 +201,8 @@ export function buildTaskPrompt(input: {
|
||||
knowledgeLines: string[];
|
||||
/** Completed runs this task already spent without reaching done. */
|
||||
priorCompletedRuns: number;
|
||||
/** What earlier sessions on this task already did, and where the last one failed. */
|
||||
journalLines: string[];
|
||||
/** Names of the tools the model can call this run. */
|
||||
availableToolNames: string[];
|
||||
/** Subtasks this task delegated; autonomously it only runs again once they are all done. */
|
||||
@@ -277,6 +279,7 @@ export function buildTaskPrompt(input: {
|
||||
`The subtasks you delegated from this task are all done: ${input.subtasks.map((subtask) => subtask.title).join(", ")}. Check their results (read_messages, list_files, list_documents) against what this task needs, then close it, or create the follow-up work that is still missing.`,
|
||||
);
|
||||
}
|
||||
lines.push(...input.journalLines);
|
||||
if (input.priorCompletedRuns > 0) {
|
||||
lines.push(
|
||||
"",
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { DatabaseConnection } from "../../src/main/database/connection";
|
||||
import { OnboardingService } from "../../src/main/services/onboarding-service";
|
||||
import { ProviderService } from "../../src/main/services/provider-service";
|
||||
import { providerAdapters } from "../../src/main/services/provider-adapters";
|
||||
import { TaskExecutionService } from "../../src/main/services/task-execution-service";
|
||||
import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker";
|
||||
import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/services/provider-tool-turn";
|
||||
|
||||
const testVault = {
|
||||
encryptSecret: (secret: string) => Buffer.from(secret, "utf8"),
|
||||
decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"),
|
||||
};
|
||||
|
||||
describe("run journal", () => {
|
||||
test("a later session on the task sees what earlier sessions did and where the last one failed", async () => {
|
||||
const prompts: string[] = [];
|
||||
let session = 0;
|
||||
const turn = async (input: ProviderToolTurnInput): Promise<AgentTurnResult> => {
|
||||
const first = input.messages[0];
|
||||
if (first?.role !== "user") throw new Error("The run must open with the task prompt.");
|
||||
if (input.messages.length === 1) {
|
||||
prompts.push(first.text);
|
||||
session += 1;
|
||||
if (session === 1) {
|
||||
return {
|
||||
text: "Working.",
|
||||
toolCalls: [
|
||||
{ id: "c1", name: "send_message", arguments: { channel: "general", body: "Kickoff: we ship the MVP." } },
|
||||
{ id: "c2", name: "read_messages", arguments: {} },
|
||||
{ id: "c3", name: "update_task_status", arguments: { task_id: "tsk_missing", status: "in_progress" } },
|
||||
],
|
||||
rawResponseJson: "{}",
|
||||
};
|
||||
}
|
||||
return { text: "Continuing.", toolCalls: [], rawResponseJson: "{}" };
|
||||
}
|
||||
throw new Error("Provider dropped the connection mid-run");
|
||||
};
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-journal-${randomUUID()}.db`));
|
||||
const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn);
|
||||
const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) });
|
||||
const execution = new TaskExecutionService(connection.db, providers);
|
||||
let orgId = "";
|
||||
connection.transaction(() => {
|
||||
orgId = onboarding.bootstrap({
|
||||
companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "",
|
||||
executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false,
|
||||
goalTitle: "Ship the MVP.", goalDescription: "A working product.",
|
||||
providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4",
|
||||
}).orgId;
|
||||
});
|
||||
const kickoff = claimNextQueuedRun(connection)!;
|
||||
await expect(execution.executeRun(orgId, kickoff.id)).rejects.toThrow("Provider dropped the connection mid-run");
|
||||
|
||||
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toHaveLength(1);
|
||||
const retry = claimNextQueuedRun(connection)!;
|
||||
await execution.executeRun(orgId, retry.id);
|
||||
|
||||
expect(prompts[0]).not.toContain("Earlier sessions on this task");
|
||||
expect(prompts[1]).toContain("Earlier sessions on this task already made these calls; each shows its result. Do not redo the ones that succeeded; retry or fix only what failed:");
|
||||
expect(prompts[1]).toContain('- update_task_status({"task_id":"tsk_missing","status":"in_progress"}) -> ERROR: Task was not found.');
|
||||
expect(prompts[1]).toMatch(/- send_message\(\{"channel":"general","body":"Kickoff: we ship the MVP\."\}\) -> /);
|
||||
expect(prompts[1]).not.toContain("- read_messages(");
|
||||
expect(prompts[1]).toContain("The last session on this task failed with: Provider dropped the connection mid-run.");
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
@@ -24,7 +24,7 @@ describe("domain services", () => {
|
||||
const connection = createConnection();
|
||||
|
||||
const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>;
|
||||
expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69]);
|
||||
expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70]);
|
||||
connection.close();
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user