diff --git a/apps/server/src/computer.ts b/apps/server/src/computer.ts index 5ea5cce96..2c9828583 100644 --- a/apps/server/src/computer.ts +++ b/apps/server/src/computer.ts @@ -27,6 +27,8 @@ import { AppError } from "./errors.ts"; const hash = (value: string) => createHash("sha256").update(value).digest("hex"); export const computerOutputLimit = 128 * 1024; const fileLimit = 256 * 1024; +// File operations that change the workspace, so an unfinished one may still have happened. +const fileWrites = new Set(["write", "write_pdf", "mkdir"]); const leaseDuration = 180000; export const computerCommandSchema = z.object({ command: z.string().trim().min(1).max(16000), @@ -611,11 +613,25 @@ export class ComputerService { const expectedOutput = operation === "read_pdf" ? maxOutputBytes : 0; const seconds = 8 + Math.ceil(Math.max(input.length, expectedOutput) / (1024 * 1024)); const result = await session.file(input, seconds, maxOutputBytes); - if (result.exitCode !== 0 || result.timedOut || result.interrupted || result.truncated) + // The in-sandbox `timeout` usually stops a slow operation first: exit 124, or 137 when + // it has to kill it. + const stopped = + result.timedOut || result.interrupted || result.exitCode === 124 || result.exitCode === 137; + if (stopped && fileWrites.has(operation)) throw new AppError( - "Computer file operation failed. Check the path, permissions and file size; symlinks cannot be opened.", + "The computer stopped before confirming this file change, so it may or may not have been saved. Check the file before trying again.", 422, ); + if (result.exitCode !== 0 || stopped || result.truncated) { + // files.py prints the reason (e.g. "File exceeds size limit") as its last stderr line. + const reason = stopped || result.truncated ? "" : result.stderr.trim().split("\n").at(-1); + throw new AppError( + reason + ? `Computer file operation failed: ${reason.trim().slice(0, 200)}` + : "Computer file operation failed. Check the path, permissions and file size; symlinks cannot be opened.", + 422, + ); + } try { return JSON.parse(result.stdout) as T; } catch { diff --git a/apps/server/src/engine/model.ts b/apps/server/src/engine/model.ts index 28c71a311..9738b6e07 100644 --- a/apps/server/src/engine/model.ts +++ b/apps/server/src/engine/model.ts @@ -363,6 +363,7 @@ export async function executeModelTask( forwardedProps: {}, }; let text = ""; + let textMessageId: string | undefined; let runError: string | undefined; await new Promise((resolve, reject) => { const timeout = setTimeout(() => { @@ -382,8 +383,13 @@ export async function executeModelTask( event.type === EventType.TEXT_MESSAGE_CONTENT) && "delta" in event && typeof event.delta === "string" - ) + ) { + // Each model step is its own message, so keep their text apart. + const messageId = "messageId" in event ? String(event.messageId) : undefined; + if (text && messageId !== textMessageId) text += "\n\n"; + textMessageId = messageId; text += event.delta; + } if (event.type === EventType.RUN_ERROR && "message" in event) runError = String(event.message); }, diff --git a/apps/server/src/engine/service.ts b/apps/server/src/engine/service.ts index c0b16dbe8..e9ad0a01f 100644 --- a/apps/server/src/engine/service.ts +++ b/apps/server/src/engine/service.ts @@ -648,12 +648,14 @@ export class AgentService { if (idea.status === "dismissed" || (idea.status === "accepted" && action === "dismiss")) return idea; if (action === "dismiss") - return this.db.compareAndSwap( - owner, - "ideas", - id, - { status: "new" }, - { status: "dismissed" }, + return ( + (await this.db.compareAndSwap( + owner, + "ideas", + id, + { status: "new" }, + { status: "dismissed" }, + )) ?? this.db.get(owner, "ideas", id) ); if (idea.status === "new") { // A task already working on (or done with) the same email is opened instead of copied. diff --git a/tests/computer.test.ts b/tests/computer.test.ts index 3ca73b504..1c870b3d6 100644 --- a/tests/computer.test.ts +++ b/tests/computer.test.ts @@ -136,6 +136,43 @@ test("paths cannot escape the workspace and file contents travel on stdin", asyn assert.ok(!call.args.some((arg) => arg.includes(text))); }); +test("file operations report why they failed and never call an unfinished write a failure", async () => { + const failing = fixture({ + owner: "file-error", + command: async () => ({ ...ok(), stderr: "File exceeds size limit\n", exitCode: 1 }), + }); + await assert.rejects( + new ComputerService(db, config, failing.runner).read("file-error", "/workspace/big.bin"), + /File exceeds size limit/, + ); + + const stopped = fixture({ + owner: "file-timeout", + command: async () => ({ ...ok(), timedOut: true, exitCode: null }), + }); + const service = new ComputerService(db, config, stopped.runner); + await assert.rejects( + service.write("file-timeout", "/workspace/note.txt", "hello"), + /may or may not have been saved/, + ); + // A read that never finished changed nothing, so it keeps the general message. + await assert.rejects(service.read("file-timeout", "/workspace/note.txt"), /Check the path/); + + // Usually the timeout inside the computer stops the operation first. + const timedOut = fixture({ + owner: "file-inner-timeout", + command: async () => ({ ...ok(), exitCode: 124 }), + }); + await assert.rejects( + new ComputerService(db, config, timedOut.runner).write( + "file-inner-timeout", + "/workspace/note.txt", + "hello", + ), + /may or may not have been saved/, + ); +}); + test("Stop interrupts an active command and its final output cannot overwrite the interrupted receipt", async () => { let finish: ((result: DockerResult) => void) | undefined; let signalStarted: (() => void) | undefined; diff --git a/tests/model-worker.test.ts b/tests/model-worker.test.ts index 1092b96ae..f4f310e04 100644 --- a/tests/model-worker.test.ts +++ b/tests/model-worker.test.ts @@ -5,6 +5,7 @@ import { createServer } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { test } from "node:test"; +import { LLMock } from "@copilotkit/aimock"; import { createApp } from "../apps/server/src/app.ts"; import { createStore, type Store } from "../apps/server/src/db.ts"; import { createDemoModel, demoModel } from "../apps/server/src/demo/model.ts"; @@ -252,6 +253,66 @@ test("the model worker keeps the text a model replies with when it calls no tool } }); +test("the model worker keeps the text of separate model steps apart", async (t) => { + const directory = await mkdtemp(join(tmpdir(), "openmuse-model-steps-")); + const db = await createStore(); + // One step says something and makes a plan, the next step replies after the plan is saved. + const mock = new LLMock({ + host: "127.0.0.1", + port: 0, + latency: 0, + strict: true, + logLevel: "silent", + }).on({ model: "openmuse-browser-demo" }, (request) => + request.messages.some((message) => message.role === "tool") + ? { content: "The plan is saved." } + : { + content: "I'll make a plan.", + toolCalls: [ + { + id: "call_plan", + name: "set_plan", + arguments: JSON.stringify({ steps: ["Pick a day"] }), + }, + ], + }, + ); + await mock.start(); + const previousBase = process.env.OPENAI_BASE_URL; + const previousKey = process.env.OPENAI_API_KEY; + process.env.OPENAI_BASE_URL = `${mock.url}/v1`; + process.env.OPENAI_API_KEY = "local-demo-test"; + t.after(async () => { + if (previousBase === undefined) delete process.env.OPENAI_BASE_URL; + else process.env.OPENAI_BASE_URL = previousBase; + if (previousKey === undefined) delete process.env.OPENAI_API_KEY; + else process.env.OPENAI_API_KEY = previousKey; + await mock.stop(); + }); + const server = await createApp(db, { + mode: "sample", + port: 8787, + host: "127.0.0.1", + publicUrl: "http://localhost:8787", + dataDir: directory, + agentBackend: "model", + intelligenceApiKey: "test-project-key-never-sent", + model: demoModel, + googleRedirectUri: "http://localhost:8787/api/google/callback", + allowedOrigins: [], + }); + try { + const task = await server.agent.createTask("owner", { prompt: "Plan my week" }); + await server.agent.worker.tick(); + const result = await server.agent.detail("owner", task.id); + assert.equal(result.task.state.lastUpdate, "I'll make a plan.\n\nThe plan is saved."); + } finally { + await server.agent.stop(); + await db.close(); + await rm(directory, { recursive: true, force: true }); + } +}); + test("replaying a completed prepared action returns its receipt without reopening approval", async (t) => { const directory = await mkdtemp(join(tmpdir(), "openmuse-model-replay-")); const db = await createStore(); diff --git a/tests/workflows.test.ts b/tests/workflows.test.ts index cb8e059af..198c41a72 100644 --- a/tests/workflows.test.ts +++ b/tests/workflows.test.ts @@ -15,6 +15,16 @@ import type { import type { ActionProposal, Artifact } from "../packages/domain/src/index.ts"; import { createSamplePdf } from "../packages/integrations/src/pdf.ts"; +function deferred() { + let resolve: (value: T) => void = () => { + throw new Error("Promise was not initialized"); + }; + const promise = new Promise((fulfill) => { + resolve = fulfill; + }); + return { promise, resolve }; +} + let db: Store, server: Awaited>, directory: string; const owner = "workflow-user"; const reconcile = () => (server.agent as unknown as { maintain(): Promise }).maintain(); @@ -374,6 +384,41 @@ test("dismissal racing acceptance never creates work for a dismissed idea", asyn } }); +test("dismissing an idea that was just accepted returns the accepted idea", async (t) => { + const idea: Idea = { + id: "race-dismiss-after-accept", + title: "Plan a picnic", + reason: "User context", + prompt: "Plan a picnic", + kind: "plan", + input: {}, + evidence: [], + status: "new", + createdAt: new Date().toISOString(), + }; + await db.put(owner, "ideas", idea); + const originalGet = db.get.bind(db); + const read = deferred(); + const resumeRead = deferred(); + let intercept = true; + t.mock.method(db, "get", async (...args: Parameters) => { + const result = await originalGet(...args); + if (intercept && args[1] === "ideas" && args[2] === idea.id) { + intercept = false; + read.resolve(); + await resumeRead.promise; + } + return result; + }); + // The dismissal reads the idea as new, then the acceptance lands before it writes. + const dismissal = server.agent.decideIdea(owner, idea.id, "dismiss"); + await read.promise; + await server.agent.decideIdea(owner, idea.id, "accept"); + resumeRead.resolve(); + const result = await dismissal; + assert.equal(result?.status, "accepted"); +}); + test("milestone-linked outcomes preserve manual progress and legacy goal tasks still append once", async () => { const goal = await server.agent.createGoal(owner, { title: "Budget",