From d84d9ac2374d9a1385609b0e4e7aca869dec28d0 Mon Sep 17 00:00:00 2001 From: "Ralph Sto. Domingo" Date: Tue, 13 Oct 2026 12:36:49 +0800 Subject: [PATCH 1/3] fix(workspace): attach the engine for a message sent while the assistant is busy Such a message joins the running prompt loop, and the engine's turn boundary ran only on a loop's first catalog. It now runs on the first catalog of each user request, counted by lesson retrieval's rule (moved into `UserRequest.requestID`), so the loop's own messages keep the turn's pin. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01HRUhVhg6XwxSKJs1iTufiA --- .../src/altimate/session/user-request.ts | 35 +++ .../src/altimate/workspace/turn-boundary.ts | 30 +++ packages/opencode/src/session/prompt.ts | 62 +++--- .../altimate/session/user-request.test.ts | 83 +++++++ .../altimate/workspace/turn-boundary.test.ts | 94 ++++++++ .../workspace-turn-boundary-loop.test.ts | 204 ++++++++++++++++++ 6 files changed, 479 insertions(+), 29 deletions(-) create mode 100644 packages/opencode/src/altimate/session/user-request.ts create mode 100644 packages/opencode/src/altimate/workspace/turn-boundary.ts create mode 100644 packages/opencode/test/altimate/session/user-request.test.ts create mode 100644 packages/opencode/test/altimate/workspace/turn-boundary.test.ts create mode 100644 packages/opencode/test/session/workspace-turn-boundary-loop.test.ts diff --git a/packages/opencode/src/altimate/session/user-request.ts b/packages/opencode/src/altimate/session/user-request.ts new file mode 100644 index 0000000000..1671ede3c6 --- /dev/null +++ b/packages/opencode/src/altimate/session/user-request.ts @@ -0,0 +1,35 @@ +// altimate_change - new file +import type { MessageV2 } from "../../session/message-v2" + +/** + * The user request a message carries, or undefined when it carries none. + * + * This is the rule lesson retrieval set in the prompt loop: a user message is a request when it has a text part that + * is neither synthetic nor ignored, or a file part. Auto-compaction can replay a user message under a fresh id. While + * lessons are delivered for the session, the replay names the request it replays (`metadata.learnOriginalMessage` on + * one of its text parts) and stays that request; otherwise it is a request of its own. + * + * The user messages the loop writes itself carry none: the summary after a subtask, the validator retry, the + * compaction marker and the continue message after compaction. + */ +export function requestID(message: MessageV2.WithParts): string | undefined { + if (message.info.role !== "user") return undefined + const asked = message.parts.some( + (part) => (part.type === "text" && !part.synthetic && !part.ignored) || part.type === "file", + ) + if (!asked) return undefined + const original = message.parts.find( + (part): part is MessageV2.TextPart => + part.type === "text" && typeof part.metadata?.learnOriginalMessage === "string", + ) + return String(original?.metadata?.learnOriginalMessage ?? message.info.id) +} + +/** The request of the newest message that carries one. */ +export function latestRequestID(messages: MessageV2.WithParts[]): string | undefined { + for (let i = messages.length - 1; i >= 0; i--) { + const id = requestID(messages[i]) + if (id !== undefined) return id + } + return undefined +} diff --git a/packages/opencode/src/altimate/workspace/turn-boundary.ts b/packages/opencode/src/altimate/workspace/turn-boundary.ts new file mode 100644 index 0000000000..d739657139 --- /dev/null +++ b/packages/opencode/src/altimate/workspace/turn-boundary.ts @@ -0,0 +1,30 @@ +// altimate_change - new file +import type { MessageV2 } from "../../session/message-v2" +import { latestRequestID } from "../session/user-request" + +/** + * Decides, at each tool catalog of one prompt loop, whether the workspace engine's turn boundary runs. + * + * It runs on the loop's first catalog, and again on the first catalog after a new user request joins the loop. A + * message sent while the assistant is busy does not start a loop of its own: the running loop picks it up. Without the + * second rule, that message would keep the engine and tool pin the loop started with, even if the engine was installed + * in between. + * + * "New user request" is `requestID` in `altimate/session/user-request.ts`, the rule lesson retrieval uses. The user + * messages the loop writes itself (subtask summary, validator retry, compaction marker and continue message) carry no + * request, so none of them move the turn to a new engine mid-task. A compaction replay stays its original request + * while lessons are delivered; otherwise it counts as new, so the boundary runs once more where the request restarts + * from the summary. The newest request in history is the one checked, so a loop-written message arriving after a + * queued request does not hide it. + */ +export function tracker(): (messages: MessageV2.WithParts[]) => boolean { + let catalogued = false + const seen = new Set() + return (messages) => { + const request = latestRequestID(messages) + const boundary = !catalogued || (request !== undefined && !seen.has(request)) + catalogued = true + if (request !== undefined) seen.add(request) + return boundary + } +} diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 554f8a1e22..00e14c8313 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -34,6 +34,7 @@ import * as WorkspaceMemory from "../altimate/workspace/memory-sync" // altimate_change start — harness-owned, frozen approved lesson delivery import { Delivery as LessonDelivery } from "../altimate/learn/delivery" import { learnEnabled } from "../altimate/learn/config" +import * as UserRequest from "../altimate/session/user-request" // altimate_change end // altimate_change start — workspace engine turn boundary, managed-key refusal, tool precedence import * as WorkspaceEngine from "../altimate/workspace/engine-overlay" @@ -43,6 +44,7 @@ import * as Awareness from "../altimate/workspace/awareness" import * as WorkspaceIdentity from "../altimate/workspace/identity" import * as PendingTurns from "../altimate/workspace/pending-turns" import * as DatamateManagerGate from "../altimate/workspace/datamate-manager-gate" +import * as TurnBoundary from "../altimate/workspace/turn-boundary" // altimate_change end import { Plugin } from "../plugin" import PROMPT_PLAN from "../session/prompt/plan.txt" @@ -670,11 +672,13 @@ export namespace SessionPrompt { let structuredOutput: unknown | undefined let step = 0 - // altimate_change start — first tool catalog of this loop. `step` counts loop - // iterations, and an iteration can `continue` before cataloguing (pending - // compaction, context overflow), so "step === 1" is not "first catalog". - let catalogued = false - // The linked-project notice for this turn, settled at its first catalog. + // altimate_change start — which tool catalogs of this loop run the workspace + // engine's turn boundary: the first, and the first after a new user request + // joins the loop. `step` counts loop iterations, and an iteration can + // `continue` before cataloguing (pending compaction, context overflow), so + // "step === 1" is not "first catalog". + const turnBoundary = TurnBoundary.tracker() + // The linked-project notice, settled at each boundary. const workspaceNotice = DatamateManagerGate.turnNotice() // altimate_change end // altimate_change start (AI-7519) — capture bootstrap start; emitted as a @@ -1212,19 +1216,17 @@ export namespace SessionPrompt { } } const user = msgs.find((msg) => msg.info.id === lastUser.id)! - const text = user.parts.filter((part): part is MessageV2.TextPart => - part.type === "text" && !part.synthetic && !part.ignored, - ) - const files = user.parts.filter((part): part is MessageV2.FilePart => part.type === "file") - if (text.length || files.length) { - // Auto-compaction can replay a user message under a fresh DB id. It - // remains the same request for retrieval and must not consume more rules. - const original = user.parts.find((part): part is MessageV2.TextPart => - part.type === "text" && typeof part.metadata?.learnOriginalMessage === "string", - ) - const requestID = String(original?.metadata?.learnOriginalMessage ?? user.info.id) + // Auto-compaction can replay a user message under a fresh DB id. It + // remains the same request for retrieval and must not consume more rules. + // The workspace engine's turn boundary counts requests by the same rule. + const requestID = UserRequest.requestID(user) + if (requestID !== undefined) { let selected = learnRequests.get(requestID) if (!selected) { + const text = user.parts.filter((part): part is MessageV2.TextPart => + part.type === "text" && !part.synthetic && !part.ignored, + ) + const files = user.parts.filter((part): part is MessageV2.FilePart => part.type === "file") const query = [...text.map((part) => part.text), ...files.map((part) => part.filename ?? "")].join("\n") selected = await lessons.prepare(sessionID, requestID, query) learnRequests.set(requestID, selected) @@ -1445,20 +1447,22 @@ export namespace SessionPrompt { { step, agent: agent.name }, sessionID, ) - // Workspace engine turn boundary, on the turn's FIRST catalog (not its first - // loop iteration — a compaction can run before any catalog): reconcile the - // bound workspace's engine (re-link, one retry on a failed handshake), settle - // this session's outcome, announce it once per verdict — then catalog the + // Workspace engine turn boundary, on the FIRST catalog of each user request + // in this loop (not its first loop iteration — a compaction can run before + // any catalog). A message sent while the assistant is busy joins this loop, + // so it is a new request here, not a new loop; a message the loop wrote + // itself is not one (see TurnBoundary). The boundary reconciles the bound + // workspace's engine (re-link, one retry on a failed handshake), settles + // this session's outcome, announces it once per verdict — then catalogs the // tools under the same per-directory lock, so another session's boundary // cannot replace the engine between this reconcile and this snapshot. The // cold engine boot happens inside MCP's own bootstrap, bounded by its - // per-server timeout. Later catalogs keep the engine tools this turn started - // with. - const firstCatalog = !catalogued - catalogued = true - const tools = firstCatalog ? await WorkspaceEngine.atTurnStart(sessionID, catalog) : await catalog() - WorkspaceEngine.pinTurnTools(sessionID, firstCatalog, tools) - if (firstCatalog) workspaceNotice.settle(sessionID) + // per-server timeout. Later catalogs keep the engine tools this request + // started with. + const boundary = turnBoundary(msgs) + const tools = boundary ? await WorkspaceEngine.atTurnStart(sessionID, catalog) : await catalog() + WorkspaceEngine.pinTurnTools(sessionID, boundary, tools) + if (boundary) workspaceNotice.settle(sessionID) // altimate_change end // Inject StructuredOutput tool if JSON schema mode enabled @@ -1644,8 +1648,8 @@ export namespace SessionPrompt { const workspaceAwareness = Awareness.systemSection(Precedence.forSession(sessionID)) // A linked session: `datamate_manager` is turned off for the project, so the // model is told that a request to connect a datamate is about this link and - // what the engine's state means for it — as settled at this turn's first - // catalog, like the tools it describes. "" for an unlinked project, so other + // what the engine's state means for it — as settled at this request's + // boundary, like the tools it describes. "" for an unlinked project, so other // sessions assemble the same array as before. const workspaceEngineNotice = workspaceNotice.text() // altimate_change end diff --git a/packages/opencode/test/altimate/session/user-request.test.ts b/packages/opencode/test/altimate/session/user-request.test.ts new file mode 100644 index 0000000000..c929dc341b --- /dev/null +++ b/packages/opencode/test/altimate/session/user-request.test.ts @@ -0,0 +1,83 @@ +// altimate_change - new file +// +// Which user messages are requests. The rule is lesson retrieval's (the prompt loop picks a request's +// lessons by it); the workspace engine's turn boundary reuses it. Each row is the shape one writer +// produces: the user, or the loop writing a message itself. +import { describe, expect, test } from "bun:test" +import type { MessageV2 } from "../../../src/session/message-v2" +import { latestRequestID, requestID } from "../../../src/altimate/session/user-request" + +let counter = 0 +const next = (prefix: string) => `${prefix}_${String(++counter).padStart(4, "0")}` + +function message(role: "user" | "assistant", parts: Record[], id = next("msg")): MessageV2.WithParts { + return { + info: { id, sessionID: "ses_test", role, time: { created: 0 } }, + parts: parts.map((part) => ({ id: next("prt"), sessionID: "ses_test", messageID: id, ...part })), + } as unknown as MessageV2.WithParts +} + +const text = (value: string, extra: Record = {}) => ({ type: "text", text: value, ...extra }) +const file = (mime: string) => ({ type: "file", mime, url: "file:///tmp/notes.txt", filename: "notes.txt" }) +const ORIGINAL = "msg_original" + +describe("requestID", () => { + const rows: [string, MessageV2.WithParts, "own" | typeof ORIGINAL | undefined][] = [ + ["a user's text", message("user", [text("connect jira")]), "own"], + [ + "a user's text beside synthetic file expansions", + message("user", [text("read this"), text("", { synthetic: true }), file("text/plain")]), + "own", + ], + ["a user's file alone", message("user", [file("text/plain")]), "own"], + [ + "the subtask summary the loop writes", + message("user", [text("Summarize the task tool output above and continue with your task.", { synthetic: true })]), + undefined, + ], + ["the validator retry the loop writes", message("user", [text("validator failures: ...", { synthetic: true })]), undefined], + ["the compaction marker", message("user", [{ type: "compaction", auto: true }]), undefined], + ["the continue message after compaction", message("user", [text("Continue if you have next steps", { synthetic: true })]), undefined], + ["a message whose only text is ignored", message("user", [text("reminder", { ignored: true })]), undefined], + [ + "a replay stamped with the request it replays", + message("user", [text("connect jira", { metadata: { learnOriginalMessage: ORIGINAL } })]), + ORIGINAL, + ], + [ + "an attachment-only replay: its file, and the stamp on an ignored carrier", + message("user", [ + file("text/plain"), + text("", { synthetic: true, ignored: true, metadata: { learnOriginalMessage: ORIGINAL } }), + ]), + ORIGINAL, + ], + ["an unstamped replay (lessons not delivered)", message("user", [text("connect jira")]), "own"], + ["an assistant message", message("assistant", [text("done")]), undefined], + ] + for (const [name, msg, expected] of rows) { + test(name, () => { + expect(requestID(msg)).toBe(expected === "own" ? msg.info.id : expected) + }) + } +}) + +describe("latestRequestID", () => { + test("is the newest message that carries a request, past the loop's own messages", () => { + const asked = message("user", [text("first")]) + const queued = message("user", [text("second")]) + const history = [ + asked, + message("assistant", [text("working")]), + queued, + message("user", [text("Summarize the task tool output above", { synthetic: true })]), + message("assistant", [text("summary")]), + ] + expect(latestRequestID(history)).toBe(queued.info.id) + }) + + test("is undefined when no message carries one", () => { + expect(latestRequestID([message("user", [{ type: "compaction", auto: true }])])).toBeUndefined() + expect(latestRequestID([])).toBeUndefined() + }) +}) diff --git a/packages/opencode/test/altimate/workspace/turn-boundary.test.ts b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts new file mode 100644 index 0000000000..3c14f2141a --- /dev/null +++ b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts @@ -0,0 +1,94 @@ +// altimate_change - new file +// +// When one prompt loop runs the workspace engine's turn boundary. Each test replays the history a +// loop sees at its successive tool catalogs: the boundary runs on the first catalog and on the first +// catalog after a new user request joins the loop, never on a message the loop wrote itself. +import { describe, expect, test } from "bun:test" +import type { MessageV2 } from "../../../src/session/message-v2" +import { tracker } from "../../../src/altimate/workspace/turn-boundary" + +let counter = 0 +const next = (prefix: string) => `${prefix}_${String(++counter).padStart(4, "0")}` + +function message(role: "user" | "assistant", parts: Record[]): MessageV2.WithParts { + const id = next("msg") + return { + info: { id, sessionID: "ses_test", role, time: { created: 0 } }, + parts: parts.map((part) => ({ id: next("prt"), sessionID: "ses_test", messageID: id, ...part })), + } as unknown as MessageV2.WithParts +} + +const asked = (value: string) => message("user", [{ type: "text", text: value }]) +const reply = () => message("assistant", [{ type: "text", text: "working" }]) +const loopWrote = (value: string) => message("user", [{ type: "text", text: value, synthetic: true }]) +const marker = () => message("user", [{ type: "compaction", auto: true }]) +const replayOf = (original: MessageV2.WithParts) => + message("user", [{ type: "text", text: "again", metadata: { learnOriginalMessage: original.info.id } }]) + +/** Runs the tracker over the history as it stood at each catalog; returns which catalogs ran the boundary. */ +function catalogs(...histories: MessageV2.WithParts[][]): boolean[] { + const boundary = tracker() + return histories.map((history) => boundary(history)) +} + +describe("workspace turn boundary", () => { + test("runs on the loop's first catalog, then not again for the same request", () => { + const a = asked("connect jira") + expect(catalogs([a], [a, reply()], [a, reply(), reply()])).toEqual([true, false, false]) + }) + + test("runs on the first catalog even when the loop resumes on a message it wrote", () => { + expect(catalogs([marker(), loopWrote("Continue if you have next steps")])).toEqual([true]) + }) + + test("does not run for the messages the loop writes itself", () => { + const a = asked("connect jira") + const h1 = [a, reply()] + const h2 = [...h1, loopWrote("Summarize the task tool output above and continue with your task.")] + const h3 = [...h2, reply(), loopWrote("validator failures: ...")] + const h4 = [...h3, reply(), marker(), reply(), loopWrote("Continue if you have next steps")] + expect(catalogs([a], h2, h3, h4)).toEqual([true, false, false, false]) + }) + + test("does not run for a stamped compaction replay of the current request", () => { + const a = asked("connect jira") + expect(catalogs([a], [a, reply(), marker(), reply(), replayOf(a)])).toEqual([true, false]) + }) + + test("runs once for a message queued while the assistant was busy", () => { + const a = asked("first") + const b = asked("now use the engine") + const h1 = [a, reply()] + const h2 = [...h1, b] + const h3 = [...h2, reply()] + expect(catalogs([a], h2, h3)).toEqual([true, true, false]) + }) + + test("runs for a queued message even when the loop writes one after it before the next catalog", () => { + const a = asked("first") + const b = asked("/review with a subtask") + const h2 = [a, reply(), b, reply(), loopWrote("Summarize the task tool output above and continue with your task.")] + expect(catalogs([a], h2)).toEqual([true, true]) + }) + + test("does not run again for a stamped replay of an earlier request", () => { + const a = asked("first") + const b = asked("second") + const h2 = [a, reply(), b] + const h3 = [...h2, reply(), marker(), reply(), replayOf(a)] + expect(catalogs([a], h2, h3)).toEqual([true, true, false]) + }) + + test("runs once more for an unstamped replay, where the request restarts from the summary", () => { + // With lessons not delivered, compaction does not stamp the replay, so it is a request of its own. + const a = asked("first") + const h2 = [a, reply(), marker(), reply(), asked("first")] + expect(catalogs([a], h2)).toEqual([true, true]) + }) + + test("each loop keeps its own record", () => { + const a = asked("first") + expect(catalogs([a], [a])).toEqual([true, false]) + expect(catalogs([a])).toEqual([true]) + }) +}) diff --git a/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts b/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts new file mode 100644 index 0000000000..e3b1af78bf --- /dev/null +++ b/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts @@ -0,0 +1,204 @@ +// altimate_change - new file +/** + * The workspace engine's turn boundary through the real prompt loop. A message sent while the + * assistant is busy joins the running loop instead of starting one; it still gets its own boundary, + * so an engine installed meanwhile attaches on it. The messages the loop writes itself (here the + * compaction marker and continue message after an overflow) do not get one. + */ +import { expect, spyOn } from "bun:test" +import { Effect, Fiber, Layer } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Session } from "@/session/session" +import { SessionPrompt } from "../../src/session/prompt" +import { SessionSummary } from "../../src/session/summary" +import { Database } from "@opencode-ai/core/database/database" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import * as WorkspaceEngine from "../../src/altimate/workspace/engine-overlay" +import { testEffect } from "../lib/effect" +import { TestLLMServer } from "../lib/llm-server" +import { provideTmpdirServerLegacy, withLegacyInstance } from "./legacy-instance" +import { LSP } from "@/lsp/lsp" +import { MCP } from "../../src/mcp" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { RuntimeFlags } from "@/effect/runtime-flags" + +const mcp = Layer.succeed( + MCP.Service, + MCP.Service.of({ + status: () => Effect.succeed({}), + clients: () => Effect.succeed({}), + tools: () => Effect.succeed({}), + prompts: () => Effect.succeed({}), + resources: () => Effect.succeed({}), + add: () => Effect.succeed({ status: { status: "disabled" as const } }), + connect: () => Effect.void, + disconnect: () => Effect.void, + remove: () => Effect.void, + getPrompt: () => Effect.succeed(undefined), + readResource: () => Effect.succeed(undefined), + startAuth: () => Effect.die("unexpected MCP auth"), + authenticate: () => Effect.die("unexpected MCP auth"), + finishAuth: () => Effect.die("unexpected MCP auth"), + removeAuth: () => Effect.void, + supportsOAuth: () => Effect.succeed(false), + hasStoredTokens: () => Effect.succeed(false), + getAuthStatus: () => Effect.succeed("not_authenticated" as const), + entry: () => Effect.succeed(undefined), + listMeta: () => Effect.succeed(undefined), + snapshot: () => Effect.succeed({ tools: {}, meta: undefined }), + }), +) + +const lsp = Layer.succeed( + LSP.Service, + LSP.Service.of({ + init: () => Effect.void, + status: () => Effect.succeed([]), + hasClients: () => Effect.succeed(false), + touchFile: () => Effect.void, + diagnostics: () => Effect.succeed({}), + hover: () => Effect.succeed(undefined), + definition: () => Effect.succeed([]), + references: () => Effect.succeed([]), + implementation: () => Effect.succeed([]), + documentSymbol: () => Effect.succeed([]), + workspaceSymbol: () => Effect.succeed([]), + prepareCallHierarchy: () => Effect.succeed([]), + incomingCalls: () => Effect.succeed([]), + outgoingCalls: () => Effect.succeed([]), + }), +) + +const root = LayerNode.group([ + SessionPrompt.node, + Session.node, + SessionProjector.node, + SessionSummary.node, + Database.node, + CrossSpawnSpawner.node, + LayerNode.make(TestLLMServer.layer, []), +]) +const it = testEffect( + LayerNode.buildLayer(root, { + replacements: [ + LayerNode.replace(MCP.node, mcp), + LayerNode.replace(LSP.node, lsp), + LayerNode.replace(RuntimeFlags.node, RuntimeFlags.layer({ experimentalEventSystem: true })), + ], + }), +) + +const providerCfg = (url: string) => ({ + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { apiKey: "test-key", baseURL: url }, + }, + }, + model: "test/test-model", +}) + +/** Counts the loop's turn boundaries, calling through to the real one. */ +const countBoundaries = Effect.acquireRelease( + Effect.sync(() => spyOn(WorkspaceEngine, "atTurnStart")), + (spy) => Effect.sync(() => spy.mockRestore()), +) + +it.live("a message sent while the assistant is busy gets its own turn boundary in the same loop", () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const boundaries = yield* countBoundaries + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const session = yield* sessions.create({ + title: "queued message", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + const gate = Promise.withResolvers() + yield* llm.hold("first answer", gate.promise) + yield* llm.text("second answer") + + const first = yield* prompt + .prompt({ sessionID: session.id, agent: "build", parts: [{ type: "text", text: "first message" }] }) + .pipe(Effect.forkChild) + yield* llm.wait(1) + expect(boundaries).toHaveBeenCalledTimes(1) + + // Resumed by the LLM server's request, this fiber has lost the instance context; restore it. + const second = yield* withLegacyInstance( + prompt.prompt({ sessionID: session.id, agent: "build", parts: [{ type: "text", text: "sent while busy" }] }), + ).pipe(Effect.forkChild) + // The second message is saved while the first answer is still on hold. + for (let i = 0; i < 100; i++) { + const messages = yield* sessions.messages({ sessionID: session.id }) + const saved = messages.some( + (msg) => msg.info.role === "user" && msg.parts.some((p) => p.type === "text" && p.text === "sent while busy"), + ) + if (saved) break + yield* Effect.sleep("20 millis") + } + gate.resolve() + yield* Fiber.join(first) + yield* Fiber.join(second) + + // One loop served both messages: the second reached the model in the same run. + const hits = (yield* llm.hits).filter((hit) => !JSON.stringify(hit.body).includes("Generate a title")) + expect(hits).toHaveLength(2) + expect(JSON.stringify(hits[1].body)).toContain("sent while busy") + // And it got a boundary of its own, so the engine is reconciled and the tools re-pinned for it. + expect(boundaries).toHaveBeenCalledTimes(2) + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) + +it.live("an overflow that compacts and continues stays one turn: one boundary", () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const boundaries = yield* countBoundaries + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const session = yield* sessions.create({ + title: "compaction", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + // Overflow -> compaction summary -> continuation: the loop writes the compaction marker and + // the continue message itself. + yield* llm.error(413, { error: { message: "request entity too large" } }) + yield* llm.text("summary of the conversation so far") + yield* llm.text("final answer after compaction") + yield* prompt.prompt({ + sessionID: session.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "long turn that overflows" }], + }) + yield* prompt.loop({ sessionID: session.id }) + + const messages = yield* sessions.messages({ sessionID: session.id }) + expect(messages.some((message) => message.parts.some((part) => part.type === "compaction"))).toBe(true) + expect((yield* llm.hits).length).toBeGreaterThanOrEqual(3) + expect(boundaries).toHaveBeenCalledTimes(1) + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) From 5c3852883221ab3fe4327eedcbc141b20a68f41b Mon Sep 17 00:00:00 2001 From: "Ralph Sto. Domingo" Date: Tue, 13 Oct 2026 13:04:37 +0800 Subject: [PATCH 2/3] fix(workspace): pick the newest user request by id, not by position After a compaction that keeps a tail, the loop's history is not in id order (`filterCompacted` places the summary after the retained tail), so a positional scan could pick the retained request and skip a message queued during the compaction. Same rule as `MessageV2.latest`. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01HRUhVhg6XwxSKJs1iTufiA --- .../opencode/src/altimate/session/user-request.ts | 13 ++++++++----- .../test/altimate/session/user-request.test.ts | 7 +++++++ .../test/altimate/workspace/turn-boundary.test.ts | 12 ++++++++++++ 3 files changed, 27 insertions(+), 5 deletions(-) diff --git a/packages/opencode/src/altimate/session/user-request.ts b/packages/opencode/src/altimate/session/user-request.ts index 1671ede3c6..3749321428 100644 --- a/packages/opencode/src/altimate/session/user-request.ts +++ b/packages/opencode/src/altimate/session/user-request.ts @@ -25,11 +25,14 @@ export function requestID(message: MessageV2.WithParts): string | undefined { return String(original?.metadata?.learnOriginalMessage ?? message.info.id) } -/** The request of the newest message that carries one. */ +/** The request of the newest message that carries one. Newest by id, not by position, like `MessageV2.latest`: + * ids are chronological, and the loop's history is not in id order after a compaction that keeps a tail + * (`filterCompacted` places the summary after the retained tail). */ export function latestRequestID(messages: MessageV2.WithParts[]): string | undefined { - for (let i = messages.length - 1; i >= 0; i--) { - const id = requestID(messages[i]) - if (id !== undefined) return id + let newest: { id: string; request: string } | undefined + for (const message of messages) { + const request = requestID(message) + if (request !== undefined && (!newest || message.info.id > newest.id)) newest = { id: message.info.id, request } } - return undefined + return newest?.request } diff --git a/packages/opencode/test/altimate/session/user-request.test.ts b/packages/opencode/test/altimate/session/user-request.test.ts index c929dc341b..827b43cb0e 100644 --- a/packages/opencode/test/altimate/session/user-request.test.ts +++ b/packages/opencode/test/altimate/session/user-request.test.ts @@ -76,6 +76,13 @@ describe("latestRequestID", () => { expect(latestRequestID(history)).toBe(queued.info.id) }) + test("is the newest by id, not by position: a compaction that keeps a tail reorders history", () => { + const retained = message("user", [text("first")]) + const queued = message("user", [text("sent during compaction")]) + // `filterCompacted` places the summary after the retained tail, so the older request comes last. + expect(latestRequestID([queued, message("assistant", [text("summary")]), retained])).toBe(queued.info.id) + }) + test("is undefined when no message carries one", () => { expect(latestRequestID([message("user", [{ type: "compaction", auto: true }])])).toBeUndefined() expect(latestRequestID([])).toBeUndefined() diff --git a/packages/opencode/test/altimate/workspace/turn-boundary.test.ts b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts index 3c14f2141a..44c4d0617e 100644 --- a/packages/opencode/test/altimate/workspace/turn-boundary.test.ts +++ b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts @@ -71,6 +71,18 @@ describe("workspace turn boundary", () => { expect(catalogs([a], h2)).toEqual([true, true]) }) + test("runs for a message queued during a compaction that keeps a tail, which reorders history", () => { + // The trace from review: request A is catalogued; compaction starts and keeps A as its tail; B + // is queued after the marker, before the summary is saved; compaction writes its continuation. + // `filterCompacted` returns [marker, B, summary, A, continuation]: A last by position, B newest by id. + const a = asked("first") + const m = marker() + const b = asked("sent during compaction") + const summary = reply() + const continuation = loopWrote("Continue if you have next steps") + expect(catalogs([a], [m, b, summary, a, continuation])).toEqual([true, true]) + }) + test("does not run again for a stamped replay of an earlier request", () => { const a = asked("first") const b = asked("second") From bed0007399dec5f4287195b0e7ab55b999f74906 Mon Sep 17 00:00:00 2001 From: "Ralph Sto. Domingo" Date: Tue, 13 Oct 2026 14:56:31 +0800 Subject: [PATCH 3/3] fix(workspace): run the boundary for any unseen user request, not only the newest A stamped compaction replay is newest by id but maps back to a request already seen, so checking only the newest could hide a message queued during the compaction. Every request in history is checked now, which also makes the order of history irrelevant. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01HRUhVhg6XwxSKJs1iTufiA --- .../src/altimate/session/user-request.ts | 12 -------- .../src/altimate/workspace/turn-boundary.ts | 17 +++++++---- .../altimate/session/user-request.test.ts | 29 +------------------ .../altimate/workspace/turn-boundary.test.ts | 8 +++++ .../workspace-turn-boundary-loop.test.ts | 12 ++++---- 5 files changed, 27 insertions(+), 51 deletions(-) diff --git a/packages/opencode/src/altimate/session/user-request.ts b/packages/opencode/src/altimate/session/user-request.ts index 3749321428..c04691cf97 100644 --- a/packages/opencode/src/altimate/session/user-request.ts +++ b/packages/opencode/src/altimate/session/user-request.ts @@ -24,15 +24,3 @@ export function requestID(message: MessageV2.WithParts): string | undefined { ) return String(original?.metadata?.learnOriginalMessage ?? message.info.id) } - -/** The request of the newest message that carries one. Newest by id, not by position, like `MessageV2.latest`: - * ids are chronological, and the loop's history is not in id order after a compaction that keeps a tail - * (`filterCompacted` places the summary after the retained tail). */ -export function latestRequestID(messages: MessageV2.WithParts[]): string | undefined { - let newest: { id: string; request: string } | undefined - for (const message of messages) { - const request = requestID(message) - if (request !== undefined && (!newest || message.info.id > newest.id)) newest = { id: message.info.id, request } - } - return newest?.request -} diff --git a/packages/opencode/src/altimate/workspace/turn-boundary.ts b/packages/opencode/src/altimate/workspace/turn-boundary.ts index d739657139..6b7c8b6011 100644 --- a/packages/opencode/src/altimate/workspace/turn-boundary.ts +++ b/packages/opencode/src/altimate/workspace/turn-boundary.ts @@ -1,6 +1,6 @@ // altimate_change - new file import type { MessageV2 } from "../../session/message-v2" -import { latestRequestID } from "../session/user-request" +import { requestID } from "../session/user-request" /** * Decides, at each tool catalog of one prompt loop, whether the workspace engine's turn boundary runs. @@ -14,17 +14,22 @@ import { latestRequestID } from "../session/user-request" * messages the loop writes itself (subtask summary, validator retry, compaction marker and continue message) carry no * request, so none of them move the turn to a new engine mid-task. A compaction replay stays its original request * while lessons are delivered; otherwise it counts as new, so the boundary runs once more where the request restarts - * from the summary. The newest request in history is the one checked, so a loop-written message arriving after a - * queued request does not hide it. + * from the summary. Every request in history is checked, not only the newest: history is not in id order after a + * compaction that keeps a tail, and a loop-written message or a replay can arrive after a queued request. */ export function tracker(): (messages: MessageV2.WithParts[]) => boolean { let catalogued = false const seen = new Set() return (messages) => { - const request = latestRequestID(messages) - const boundary = !catalogued || (request !== undefined && !seen.has(request)) + let unseen = false + for (const message of messages) { + const request = requestID(message) + if (request === undefined || seen.has(request)) continue + seen.add(request) + unseen = true + } + const boundary = !catalogued || unseen catalogued = true - if (request !== undefined) seen.add(request) return boundary } } diff --git a/packages/opencode/test/altimate/session/user-request.test.ts b/packages/opencode/test/altimate/session/user-request.test.ts index 827b43cb0e..22b61a6298 100644 --- a/packages/opencode/test/altimate/session/user-request.test.ts +++ b/packages/opencode/test/altimate/session/user-request.test.ts @@ -5,7 +5,7 @@ // produces: the user, or the loop writing a message itself. import { describe, expect, test } from "bun:test" import type { MessageV2 } from "../../../src/session/message-v2" -import { latestRequestID, requestID } from "../../../src/altimate/session/user-request" +import { requestID } from "../../../src/altimate/session/user-request" let counter = 0 const next = (prefix: string) => `${prefix}_${String(++counter).padStart(4, "0")}` @@ -61,30 +61,3 @@ describe("requestID", () => { }) } }) - -describe("latestRequestID", () => { - test("is the newest message that carries a request, past the loop's own messages", () => { - const asked = message("user", [text("first")]) - const queued = message("user", [text("second")]) - const history = [ - asked, - message("assistant", [text("working")]), - queued, - message("user", [text("Summarize the task tool output above", { synthetic: true })]), - message("assistant", [text("summary")]), - ] - expect(latestRequestID(history)).toBe(queued.info.id) - }) - - test("is the newest by id, not by position: a compaction that keeps a tail reorders history", () => { - const retained = message("user", [text("first")]) - const queued = message("user", [text("sent during compaction")]) - // `filterCompacted` places the summary after the retained tail, so the older request comes last. - expect(latestRequestID([queued, message("assistant", [text("summary")]), retained])).toBe(queued.info.id) - }) - - test("is undefined when no message carries one", () => { - expect(latestRequestID([message("user", [{ type: "compaction", auto: true }])])).toBeUndefined() - expect(latestRequestID([])).toBeUndefined() - }) -}) diff --git a/packages/opencode/test/altimate/workspace/turn-boundary.test.ts b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts index 44c4d0617e..17401ac0bf 100644 --- a/packages/opencode/test/altimate/workspace/turn-boundary.test.ts +++ b/packages/opencode/test/altimate/workspace/turn-boundary.test.ts @@ -83,6 +83,14 @@ describe("workspace turn boundary", () => { expect(catalogs([a], [m, b, summary, a, continuation])).toEqual([true, true]) }) + test("runs for a queued message that a stamped replay arrives after", () => { + // From review: A is catalogued; B is queued while a compaction runs, and the compaction then + // writes a stamped replay of A. The replay is newest by id but maps back to A, already seen. + const a = asked("first") + const b = asked("sent during compaction") + expect(catalogs([a], [a, reply(), marker(), b, reply(), replayOf(a)])).toEqual([true, true]) + }) + test("does not run again for a stamped replay of an earlier request", () => { const a = asked("first") const b = asked("second") diff --git a/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts b/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts index e3b1af78bf..e3d2a37c68 100644 --- a/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts +++ b/packages/opencode/test/session/workspace-turn-boundary-loop.test.ts @@ -145,15 +145,17 @@ it.live("a message sent while the assistant is busy gets its own turn boundary i const second = yield* withLegacyInstance( prompt.prompt({ sessionID: session.id, agent: "build", parts: [{ type: "text", text: "sent while busy" }] }), ).pipe(Effect.forkChild) - // The second message is saved while the first answer is still on hold. - for (let i = 0; i < 100; i++) { + // The second message is saved while the first answer is still on hold. Asserted, not just + // awaited: released early, it could start a loop of its own and get a boundary that way. + let saved = false + for (let i = 0; i < 100 && !saved; i++) { const messages = yield* sessions.messages({ sessionID: session.id }) - const saved = messages.some( + saved = messages.some( (msg) => msg.info.role === "user" && msg.parts.some((p) => p.type === "text" && p.text === "sent while busy"), ) - if (saved) break - yield* Effect.sleep("20 millis") + if (!saved) yield* Effect.sleep("20 millis") } + expect(saved).toBe(true) gate.resolve() yield* Fiber.join(first) yield* Fiber.join(second)