Skip to content

ACP changes for DocuMentor's queued messages: messageId, empty-prompt resume, _meta.opencode #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheFabFab merged 4 commits into from
Oct 9, 2026
Merged

ACP changes for DocuMentor's queued messages: messageId, empty-prompt resume, _meta.opencode #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4383134
feat(acp): store a prompt under the client's msg-prefixed message id
TheFabFab Oct 8, 2026
27b3dc1
feat(session): resume the loop on a prompt with no parts
TheFabFab Oct 8, 2026
390396d
feat(acp): name the assistant message and its parent on every content…
TheFabFab Oct 8, 2026
caadf43
fix(acp): look up a message's parent once, failures included, and onl…
TheFabFab Oct 8, 2026
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
41 changes: 24 additions & 17 deletions documentor/FORK.md
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions documentor/upstream-test-exceptions.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,6 @@ make it pass on Postgres or keep it listed here.
| `packages/core/test/integration.test.ts`: "completes auto OAuth in the background" | Changed | It yielded once and then expected the attempt to be complete. The test now waits for the attempt to leave "pending" and then asserts, with nothing in between, that it is complete and its credential is listed | — |
| `packages/sdk-next/test/embedded.test.ts` | Changed | Each test pointed `Flag.OPENCODE_DB` at a temporary SQLite file. The file sets the throwaway-schema environment once instead | — |
| `packages/opencode/test/lib/cli-process.ts` and `test/cli/run/run-process.test.ts` | Changed | The subprocess harness allowed 30 s per `opencode run`, and the file 60 s per test. Thirteen subprocesses compile the CLI from source at once on a four-core runner; on unmodified upstream three of them already exceed 30 s there, and each process here also opens a database schema. Now 90 s and 120 s; the "exits nonzero promptly" case allows 45 s instead of 15 s and still proves the process exits rather than hanging until the harness kills it | — |
| `packages/opencode/test/acp/event.test.ts`: "uses reasoning part ids as ACP thought message boundaries" | Changed | Each expected update also carries `_meta.opencode` (FORK.md, `acp/event.ts`) | — |
| `packages/opencode/test/acp/service-session.test.ts`: "replays loaded session transcript chunks", "replays reasoning parts as separate ACP thought messages" | Changed | Each expected assistant or thought update also carries `_meta.opencode` (FORK.md, `acp/event.ts`) | — |
| `packages/opencode/test/acp/event.test.ts`: "does not call sdk.session.message repeatedly when metadata is known" | Changed | Every content update also names the message's parent, so "metadata is known" includes the parent. The test delivers the message's `message.updated` before its deltas, as opencode does, and still asserts no lookups and five updates (FORK.md, `acp/event.ts`) | — |
88 changes: 84 additions & 4 deletions packages/opencode/src/acp/event.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import type {
Event,
EventMessagePartDelta,
EventMessagePartUpdated,
Message,
OpencodeClient,
Part,
SessionMessageResponse,
Expand Down Expand Up @@ -30,6 +31,14 @@ type GlobalEventStream = {
stream: AsyncIterable<GlobalEventEnvelope>
}

/**
* `_meta.opencode` on every content and tool update: the assistant message the
* update belongs to, and the user message that assistant answers. A client that
* sends a prompt while a turn runs reads from it the moment the agent starts
* answering that prompt.
*/
export type OpencodeUpdateMeta = { opencode: { messageId: string; parentId: string | null } }

export function start(input: { sdk: OpencodeClient; connection: Connection; session: ACPSession.Interface }) {
const subscription = new Subscription(input)
subscription.start()
Expand All @@ -42,6 +51,10 @@ export class Subscription {
private readonly toolStarts = new Set<string>()
private readonly connectionWaiters = new Set<() => void>()
private readonly idleWaiters = new Map<string, Set<ReturnType<typeof signal>>>()
// Per tracked session: assistant message id → its parentID. Filled from message.updated, which opencode publishes
// before any part of the message, or by one lookup on a miss; a failed lookup holds null. Forgotten when the
// session goes idle and when the event stream disconnects.
private readonly messageParents = new Map<string, Map<string, Promise<string | null>>>()
private readonly permission: ACPPermission.Handler
private connected = false
private started = false
Expand Down Expand Up @@ -95,6 +108,8 @@ export class Subscription {
case "session.status":
if (event.properties.status.type === "idle") this.idle(event.properties.sessionID)
return
case "message.updated":
return this.handleMessageUpdated(event.properties.info)
case "permission.asked":
this.permission.handle(event)
return
Expand All @@ -107,6 +122,7 @@ export class Subscription {

async replayMessage(message: SessionMessageResponse) {
if (message.info.role !== "assistant" && message.info.role !== "user") return
this.recordMessage(message.info)

const cwd = message.info.role === "assistant" ? message.info.path?.cwd : undefined
for (const part of message.parts) {
Expand Down Expand Up @@ -136,6 +152,16 @@ export class Subscription {
sessionUpdate,
messageId: part.type === "reasoning" ? part.id : message.info.id,
...chunk,
...(sessionUpdate === "user_message_chunk"
? {}
: {
_meta: {
opencode: {
messageId: message.info.id,
parentId: message.info.role === "assistant" ? (message.info.parentID ?? null) : null,
},
} satisfies OpencodeUpdateMeta,
}),
},
})
}
Expand Down Expand Up @@ -172,6 +198,7 @@ export class Subscription {
}

private disconnected() {
this.messageParents.clear()
if (!this.connected) return
this.connected = false
const error = new Error("ACP event stream disconnected")
Expand All @@ -182,12 +209,56 @@ export class Subscription {
}

private idle(sessionId: string) {
this.messageParents.delete(sessionId)
const waiters = this.idleWaiters.get(sessionId)
if (!waiters) return
this.idleWaiters.delete(sessionId)
for (const waiter of waiters) waiter.resolve()
}

private async handleMessageUpdated(info: Message) {
const session = await Effect.runPromise(this.input.session.tryGet(info.sessionID))
if (!session) return
this.recordMessage(info)
}

// Only assistant messages: updates name the assistant message they belong to, never a user message.
private recordMessage(info: Message) {
if (info.role !== "assistant") return
this.sessionParents(info.sessionID).set(info.id, Promise.resolve(info.parentID ?? null))
}

private sessionParents(sessionId: string) {
const existing = this.messageParents.get(sessionId)
if (existing) return existing
const parents = new Map<string, Promise<string | null>>()
this.messageParents.set(sessionId, parents)
return parents
}

private async messageMeta(sessionId: string, cwd: string, messageId: string): Promise<OpencodeUpdateMeta> {
const parents = this.sessionParents(sessionId)
const known = parents.get(messageId)
if (known) return { opencode: { messageId, parentId: await known } }
// The subscription misses message.updated when it connects mid-turn or after the session went idle. One lookup
// per message answers every update for it, concurrent ones included. A failed lookup is remembered as null until
// the session goes idle, so a server that cannot answer is not asked again for each chunk of the stream.
const lookup = Promise.resolve()
.then(() =>
this.input.sdk.session.message(
{ sessionID: sessionId, messageID: messageId, directory: cwd },
{ throwOnError: true },
),
)
.then((response) => {
const info = response.data?.info
return info?.role === "assistant" ? (info.parentID ?? null) : null
})
.catch(() => null)
parents.set(messageId, lookup)
return { opencode: { messageId, parentId: await lookup } }
}

private async handlePartUpdated(event: EventMessagePartUpdated) {
const part = event.properties.part
const sessionId = part.sessionID || event.properties.sessionID
Expand Down Expand Up @@ -238,6 +309,7 @@ export class Subscription {
type: "text",
text: props.delta,
},
_meta: await this.messageMeta(session.id, session.cwd, props.messageID),
},
})
return
Expand All @@ -253,6 +325,7 @@ export class Subscription {
type: "text",
text: props.delta,
},
_meta: await this.messageMeta(session.id, session.cwd, props.messageID),
},
})
}
Expand All @@ -271,6 +344,7 @@ export class Subscription {
.then((response) => response.data)
.catch(() => undefined)
if (!message) return
this.recordMessage(message.info)

const part = message.parts.find((item) => item.id === partId)
if (!part) return
Expand All @@ -293,15 +367,16 @@ export class Subscription {
}

private async handleToolPart(sessionId: string, part: ToolPart, cwd: string) {
await this.toolStart(sessionId, part, cwd)
const meta = await this.messageMeta(sessionId, cwd, part.messageID)
await this.toolStart(sessionId, part, cwd, meta)

switch (part.state.status) {
case "pending":
this.shellSnapshots.delete(part.callID)
return

case "running":
await this.runningTool(sessionId, part, cwd)
await this.runningTool(sessionId, part, cwd, meta)
return

case "completed":
Expand All @@ -316,6 +391,7 @@ export class Subscription {
state: part.state,
cwd,
}),
_meta: meta,
},
})
return
Expand All @@ -332,13 +408,14 @@ export class Subscription {
state: part.state,
cwd,
}),
_meta: meta,
},
})
return
}
}

private async runningTool(sessionId: string, part: ToolPart, cwd: string) {
private async runningTool(sessionId: string, part: ToolPart, cwd: string, meta: OpencodeUpdateMeta) {
if (part.state.status !== "running") return

const output = part.tool === "bash" ? shellOutputSnapshot(part.state) : undefined
Expand All @@ -354,6 +431,7 @@ export class Subscription {
state: part.state,
cwd,
}),
_meta: meta,
},
})
return
Expand All @@ -372,11 +450,12 @@ export class Subscription {
output,
cwd,
}),
_meta: meta,
},
})
}

private async toolStart(sessionId: string, part: ToolPart, cwd: string) {
private async toolStart(sessionId: string, part: ToolPart, cwd: string, meta: OpencodeUpdateMeta) {
if (this.toolStarts.has(part.callID)) return
this.toolStarts.add(part.callID)
await this.input.connection.sessionUpdate({
Expand All @@ -389,6 +468,7 @@ export class Subscription {
state: part.state,
cwd,
}),
_meta: meta,
},
})
}
Expand Down
10 changes: 10 additions & 0 deletions packages/opencode/src/acp/service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -517,6 +517,7 @@ export function make(input: {
const modeId = current.modeId ?? (snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined)
const parts = promptContentToParts(params.prompt)
const command = detectSlashCommand(parts)
const messageID = parts.length > 0 ? opencodeMessageID(params.messageId) : undefined

if (!command) {
const response = yield* request(
Expand All @@ -525,6 +526,7 @@ export function make(input: {
input.sdk.session.prompt(
{
sessionID: current.id,
...(messageID ? { messageID } : {}),
model: {
providerID: selected.providerID,
modelID: selected.modelID,
Expand All @@ -551,6 +553,7 @@ export function make(input: {
input.sdk.session.command(
{
sessionID: current.id,
...(messageID ? { messageID } : {}),
command: known.name,
arguments: command.args,
model: `${selected.providerID}/${selected.modelID}`,
Expand Down Expand Up @@ -836,6 +839,13 @@ function detectSlashCommand(parts: ReturnType<typeof promptContentToParts>) {
return { name, args: rest.join(" ").trim() }
}

// A client may name the user message its prompt creates, so that its own copy and opencode's share one id.
// Only an id that is already an opencode message id (prefix "msg") is used; ACP clients that send UUIDs get an
// opencode-minted id. An empty prompt creates no message (SessionPrompt.prompt), so it names none.
function opencodeMessageID(messageId: string | null | undefined) {
return typeof messageId === "string" && messageId.startsWith("msg") ? messageId : undefined
}

const promptResponse = Effect.fn("ACP.promptResponse")(function* (
info: AssistantInfo,
messageId: string | null | undefined,
Expand Down
11 changes: 11 additions & 0 deletions packages/opencode/src/session/prompt.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1054,6 +1054,12 @@ const layer = Layer.effect(
)(function* (input: PromptInput) {
const session = yield* sessions.get(input.sessionID).pipe(Effect.orDie)
yield* revert.cleanup(session)
if (input.parts.length === 0 && input.noReply !== true && (yield* hasUserMessage(input.sessionID))) {
// A prompt with no parts resumes the session instead of storing an empty user message. The loop runs over
// the messages already stored, so user messages that arrived during a cancelled step are answered, and a
// session with nothing left to answer returns its last assistant without calling the model.
return yield* loop({ sessionID: input.sessionID })
}
const message = yield* createUserMessage(input)
yield* sessions.touch(input.sessionID)

Expand All @@ -1078,6 +1084,11 @@ const layer = Layer.effect(
throw new Error("Impossible")
})

const hasUserMessage = Effect.fnUntraced(function* (sessionID: SessionID) {
const match = yield* sessions.findMessage(sessionID, (m) => m.info.role === "user").pipe(Effect.orDie)
return Option.isSome(match)
})

const runLoop: (sessionID: SessionID) => Effect.Effect<SessionV1.WithParts> = Effect.fn("SessionPrompt.run")(
function* (sessionID: SessionID) {
const ctx = yield* InstanceState.context
Expand Down
Loading
Loading