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
26 changes: 26 additions & 0 deletions packages/opencode/src/altimate/session/user-request.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
// 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",

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

WARNING: Count subtask-only commands as user requests for the turn boundary

The built-in /review command has subtask: true (command/index.ts) and SessionPrompt.command() saves it as a user message containing only a subtask part. If it is queued while request A is busy, the running loop executes the subtask and then writes a synthetic summary; neither message passes this text-or-file check. latestRequestID() still returns A, so the next catalog skips atTurnStart() and the command's parent continuation keeps A's engine and tool pin, even when an engine was installed between requests. The tracker needs to recognize a user-authored subtask as a new boundary without treating the loop-written summary as one; the queued-subtask test currently models /review as ordinary text.


Reply with @kilocode-bot fix it to have Kilo Code address this issue.

)
if (!asked) return undefined
const original = message.parts.find(
(part): part is MessageV2.TextPart =>
part.type === "text" && typeof part.metadata?.learnOriginalMessage === "string",
Comment thread
ralphstodomingo marked this conversation as resolved.
)
return String(original?.metadata?.learnOriginalMessage ?? message.info.id)
}
35 changes: 35 additions & 0 deletions packages/opencode/src/altimate/workspace/turn-boundary.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
// altimate_change - new file
import type { MessageV2 } from "../../session/message-v2"
import { requestID } 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. 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<string>()
return (messages) => {
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
return boundary
}
}
62 changes: 33 additions & 29 deletions packages/opencode/src/session/prompt.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
63 changes: 63 additions & 0 deletions packages/opencode/test/altimate/session/user-request.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
// 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 { 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<string, unknown>[], 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<string, unknown> = {}) => ({ 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("<file contents>", { 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)
})
}
})
114 changes: 114 additions & 0 deletions packages/opencode/test/altimate/workspace/turn-boundary.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
// 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<string, unknown>[]): 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("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("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")
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])
})
})
Loading
Loading