authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-06 02:15:53-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 02:15:53-07:00
log281b92b48810062275798dd4107364384ebaf432
treec1a2bf2156f83a7095fb033f65b6877730854e81
parent60e6fa30c3a8c1e17ba7fb80e5f85f44a0cd2693
signature Signed by PGP key B5690EEEBB952194

perf(server): stop scanning old OpenCode parts (#10116)

2 files changed, 411 insertions(+), 81 deletions(-)

apps/server/src/provider/Layers/OpenCodeAdapter.test.ts+308-1
...@@ -16,7 +16,7 @@ import * as Schema from "effect/Schema";...@@ -16,7 +16,7 @@ import * as Schema from "effect/Schema";
16import * as Scope from "effect/Scope";16import * as Scope from "effect/Scope";
17import * as Stream from "effect/Stream";17import * as Stream from "effect/Stream";
18import * as TestClock from "effect/testing/TestClock";18import * as TestClock from "effect/testing/TestClock";
19import { beforeEach } from "vite-plus/test";19import { beforeEach, vi } from "vite-plus/test";
20import type {20import type {
21 Event as OpenCodeEvent,21 Event as OpenCodeEvent,
22 PermissionRequest,22 PermissionRequest,
...@@ -578,6 +578,18 @@ function promiseWithResolvers<T>() {...@@ -578,6 +578,18 @@ function promiseWithResolvers<T>() {
578 return { promise, resolve, reject };578 return { promise, resolve, reject };
579}579}
580580
581function makeOpenCodeEventQueue() {
582 let pending = promiseWithResolvers<unknown>();
583 const events = [pending.promise];
584 runtimeMock.state.subscribedEvents = events;
585 return (event: unknown) => {
586 const current = pending;
587 pending = promiseWithResolvers<unknown>();
588 events.push(pending.promise);
589 current.resolve(event);
590 };
591}
592
581const permissionRequest = (id: string, sessionID: string): PermissionRequest => ({593const permissionRequest = (id: string, sessionID: string): PermissionRequest => ({
582 id,594 id,
583 sessionID,595 sessionID,
...@@ -6708,6 +6720,301 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {...@@ -6708,6 +6720,301 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
6708 }),6720 }),
6709 );6721 );
67106722
6723 it.effect("processes late assistant metadata without visiting completed turns", () =>
6724 Effect.gen(function* () {
6725 const adapter = yield* OpenCodeAdapter;
6726 const threadId = asThreadId("thread-indexed-opencode-parts");
6727 const sessionID = "http://127.0.0.1:9999/session";
6728 const enqueue = makeOpenCodeEventQueue();
6729 yield* adapter.startSession({
6730 provider: ProviderDriverKind.make("opencode"),
6731 threadId,
6732 runtimeMode: "full-access",
6733 });
6734
6735 for (let index = 0; index < 24; index += 1) {
6736 const completed = yield* adapter.streamEvents.pipe(
6737 Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"),
6738 Stream.runHead,
6739 Effect.forkChild,
6740 );
6741 yield* adapter.sendTurn({
6742 threadId,
6743 input: `Complete turn ${index}`,
6744 modelSelection: createModelSelection(
6745 ProviderInstanceId.make("opencode"),
6746 "opencode/kimi-k3",
6747 ),
6748 });
6749 enqueue({
6750 type: "message.updated",
6751 properties: { sessionID, info: { id: `history-message-${index}`, role: "assistant" } },
6752 });
6753 enqueue({
6754 type: "message.part.updated",
6755 properties: {
6756 sessionID,
6757 part: {
6758 id: `history-part-${index}`,
6759 messageID: `history-message-${index}`,
6760 sessionID,
6761 type: "text",
6762 text: `Completed turn ${index}`,
6763 time: { start: 1, end: 2 },
6764 },
6765 },
6766 });
6767 enqueue({
6768 type: "session.status",
6769 properties: { sessionID, status: { type: "idle" } },
6770 });
6771 yield* Fiber.join(completed);
6772 }
6773
6774 let visitedHistoryParts = 0;
6775 const values = Map.prototype.values;
6776 yield* Effect.acquireRelease(
6777 Effect.sync(() =>
6778 vi
6779 .spyOn(Map.prototype, "values")
6780 .mockImplementation(function (this: Map<unknown, unknown>) {
6781 const iterator = values.call(this);
6782 const next = iterator.next.bind(iterator);
6783 iterator.next = () => {
6784 const result = next();
6785 const value: unknown = result.value;
6786 if (
6787 typeof value === "object" &&
6788 value !== null &&
6789 "id" in value &&
6790 typeof value.id === "string" &&
6791 value.id.startsWith("history-part-")
6792 ) {
6793 visitedHistoryParts += 1;
6794 }
6795 return result;
6796 };
6797 return iterator;
6798 }),
6799 ),
6800 (spy) => Effect.sync(() => spy.mockRestore()),
6801 );
6802
6803 const stepProcessed = yield* Deferred.make<void>();
6804 const eventsFiber = yield* adapter.streamEvents.pipe(
6805 Stream.filter((event) => event.threadId === threadId),
6806 Stream.tap((event) =>
6807 event.type === "thread.state.changed"
6808 ? Deferred.succeed(stepProcessed, undefined)
6809 : Effect.void,
6810 ),
6811 Stream.takeUntil((event) => event.type === "turn.completed"),
6812 Stream.runCollect,
6813 Effect.forkChild,
6814 );
6815 yield* adapter.sendTurn({
6816 threadId,
6817 input: "Process late metadata",
6818 modelSelection: createModelSelection(
6819 ProviderInstanceId.make("opencode"),
6820 "opencode/kimi-k3",
6821 ),
6822 });
6823 const promptMessageId = (runtimeMock.state.promptCalls.at(-1) as { messageID: string })
6824 .messageID;
6825 const part = {
6826 id: "current-part",
6827 messageID: "current-message",
6828 sessionID,
6829 type: "text",
6830 text: "Current response",
6831 time: { start: 3, end: 4 },
6832 };
6833 const step = {
6834 id: "current-step",
6835 messageID: "current-message",
6836 sessionID,
6837 type: "step-finish",
6838 reason: "stop",
6839 cost: 0,
6840 tokens: { input: 40, output: 10, reasoning: 2, cache: { read: 5, write: 1 } },
6841 };
6842 enqueue({ type: "message.part.updated", properties: { sessionID, part } });
6843 enqueue({ type: "message.part.updated", properties: { sessionID, part: step } });
6844 enqueue({ type: "session.compacted", properties: { sessionID } });
6845 yield* Deferred.await(stepProcessed);
6846 yield* adapter.sendTurn({
6847 threadId,
6848 input: "Steer before metadata arrives",
6849 modelSelection: createModelSelection(
6850 ProviderInstanceId.make("opencode"),
6851 "opencode/kimi-k3",
6852 ),
6853 });
6854 for (const parentID of ["", promptMessageId]) {
6855 enqueue({
6856 type: "message.updated",
6857 properties: { sessionID, info: { id: "current-message", role: "assistant", parentID } },
6858 });
6859 }
6860 enqueue({ type: "message.part.updated", properties: { sessionID, part } });
6861 enqueue({ type: "message.part.updated", properties: { sessionID, part: step } });
6862 enqueue({
6863 type: "message.part.updated",
6864 properties: {
6865 sessionID,
6866 part: {
6867 id: "history-part-0",
6868 messageID: "history-message-0",
6869 sessionID,
6870 type: "text",
6871 text: "Completed turn zero",
6872 time: { start: 1, end: 2 },
6873 },
6874 },
6875 });
6876 enqueue({ type: "session.status", properties: { sessionID, status: { type: "idle" } } });
6877
6878 const events = yield* Fiber.join(eventsFiber);
6879 NodeAssert.equal(visitedHistoryParts, 0);
6880 NodeAssert.deepEqual(
6881 events
6882 .filter((event) => event.type === "content.delta")
6883 .map((event) => event.payload.delta),
6884 ["Current response", "zero"],
6885 );
6886 NodeAssert.equal(events.filter((event) => event.type === "item.completed").length, 1);
6887 const completed = events.find((event) => event.type === "turn.completed");
6888 NodeAssert.deepEqual(completed?.payload.tokenUsage, {
6889 usageStatus: "complete",
6890 usageScope: "main_agent",
6891 inputTokens: 46,
6892 cachedInputTokens: 5,
6893 cacheCreationTokens: 1,
6894 outputTokens: 12,
6895 reasoningTokens: 2,
6896 hasSubagents: false,
6897 });
6898 yield* adapter.stopSession(threadId);
6899 }).pipe(Effect.scoped),
6900 );
6901
6902 it.effect("keeps completed text edits and clears removed parts across reconnects", () =>
6903 Effect.gen(function* () {
6904 const adapter = yield* OpenCodeAdapter;
6905 const threadId = asThreadId("thread-opencode-text-retention");
6906 const sessionID = "http://127.0.0.1:9999/session";
6907 const messageID = "retained-message";
6908 const metadata = {
6909 type: "message.updated",
6910 properties: {
6911 sessionID,
6912 info: { id: messageID, role: "assistant", time: { created: 1, completed: 2 } },
6913 },
6914 };
6915 const snapshot = (
6916 text: string,
6917 id = "retained-part",
6918 type: "text" | "reasoning" = "text",
6919 ) => ({
6920 type: "message.part.updated",
6921 properties: {
6922 sessionID,
6923 part: { id, sessionID, messageID, type, text, time: { start: 1, end: 2 } },
6924 },
6925 });
6926 const delta = (text: string) => ({
6927 type: "message.part.delta",
6928 properties: { sessionID, messageID, partID: "retained-part", field: "text", delta: text },
6929 });
6930 const nonTextReplacement = {
6931 type: "message.part.updated",
6932 properties: {
6933 sessionID,
6934 part: {
6935 id: "retained-part",
6936 sessionID,
6937 messageID,
6938 type: "file",
6939 mime: "text/plain",
6940 url: "file:///repo/result.txt",
6941 },
6942 },
6943 };
6944 runtimeMock.state.subscribedEvents = [
6945 snapshot("Replaced before metadata"),
6946 nonTextReplacement,
6947 metadata,
6948 snapshot("Thinking", "reasoning-part", "reasoning"),
6949 snapshot("Hello world"),
6950 { type: "server.connected", properties: {} },
6951 metadata,
6952 snapshot("Thinking", "reasoning-part", "reasoning"),
6953 snapshot("Thinking more", "reasoning-part", "reasoning"),
6954 snapshot("Hello world"),
6955 snapshot("Hello"),
6956 snapshot("Hello there"),
6957 delta(" again"),
6958 snapshot("Hello there again"),
6959 nonTextReplacement,
6960 delta("ignored while file"),
6961 metadata,
6962 snapshot("Hello there again!"),
6963 {
6964 type: "message.part.removed",
6965 properties: { sessionID, messageID, partID: "retained-part" },
6966 },
6967 delta("removed part"),
6968 metadata,
6969 snapshot("Fresh"),
6970 snapshot("Second", "second-part"),
6971 { type: "message.removed", properties: { sessionID, messageID } },
6972 delta("removed message"),
6973 metadata,
6974 snapshot("New thoughts", "reasoning-part", "reasoning"),
6975 snapshot("New"),
6976 { type: "session.compacted", properties: { sessionID } },
6977 ];
6978 const eventsFiber = yield* adapter.streamEvents.pipe(
6979 Stream.filter((event) => event.threadId === threadId),
6980 Stream.takeUntil((event) => event.type === "thread.state.changed"),
6981 Stream.runCollect,
6982 Effect.forkChild,
6983 );
6984 yield* adapter.startSession({
6985 provider: ProviderDriverKind.make("opencode"),
6986 threadId,
6987 runtimeMode: "full-access",
6988 });
6989
6990 const events = yield* Fiber.join(eventsFiber);
6991 NodeAssert.deepEqual(
6992 events
6993 .filter((event) => event.type === "content.delta")
6994 .map((event) => [event.payload.streamKind, event.payload.delta]),
6995 [
6996 ["reasoning_text", "Thinking"],
6997 ["assistant_text", "Hello world"],
6998 ["reasoning_text", " more"],
6999 ["assistant_text", "there"],
7000 ["assistant_text", " again"],
7001 ["assistant_text", "!"],
7002 ["assistant_text", "Fresh"],
7003 ["assistant_text", "Second"],
7004 ["reasoning_text", "New thoughts"],
7005 ["assistant_text", "New"],
7006 ],
7007 );
7008 NodeAssert.deepEqual(
7009 events
7010 .filter((event) => event.type === "item.completed")
7011 .map((event) => event.payload.detail),
7012 ["Hello world", "Fresh", "Second", "New"],
7013 );
7014 yield* adapter.stopSession(threadId);
7015 }),
7016 );
7017
6711 it.effect("maps native task progress only while a turn is active", () =>7018 it.effect("maps native task progress only while a turn is active", () =>
6712 Effect.gen(function* () {7019 Effect.gen(function* () {
6713 const adapter = yield* OpenCodeAdapter;7020 const adapter = yield* OpenCodeAdapter;
apps/server/src/provider/Layers/OpenCodeAdapter.ts+103-80
...@@ -322,6 +322,16 @@ function isOpenCodeDefaultTitle(title: string): boolean {...@@ -322,6 +322,16 @@ function isOpenCodeDefaultTitle(title: string): boolean {
322 return OPENCODE_DEFAULT_TITLE_PATTERN.test(title);322 return OPENCODE_DEFAULT_TITLE_PATTERN.test(title);
323}323}
324324
325type OpenCodeTextPart = Extract<Part, { readonly type: "text" | "reasoning" }>;
326
327type OpenCodeTextPartState = Pick<OpenCodeTextPart, "id" | "messageID" | "type" | "time"> & {
328 text: string | undefined;
329 emittedText: string | undefined;
330 completed: boolean;
331};
332
333type OpenCodeStepUsage = Pick<Extract<Part, { readonly type: "step-finish" }>, "id" | "tokens">;
334
325interface OpenCodeSessionContext {335interface OpenCodeSessionContext {
326 session: ProviderSession;336 session: ProviderSession;
327 readonly client: OpencodeClient;337 readonly client: OpencodeClient;
...@@ -336,9 +346,9 @@ interface OpenCodeSessionContext {...@@ -336,9 +346,9 @@ interface OpenCodeSessionContext {
336 readonly pendingPermissions: Map<string, PermissionRequest>;346 readonly pendingPermissions: Map<string, PermissionRequest>;
337 readonly pendingQuestions: Map<string, QuestionRequest>;347 readonly pendingQuestions: Map<string, QuestionRequest>;
338 readonly messageRoleById: Map<string, "user" | "assistant">;348 readonly messageRoleById: Map<string, "user" | "assistant">;
339 readonly partById: Map<string, Part>;349 // OpenCode permits edits to completed parts. Keep text for snapshot comparison
340 readonly emittedTextByPartId: Map<string, string>;350 // until native removal or session teardown, but do not retain other part payloads.
341 readonly completedAssistantPartIds: Set<string>;351 readonly textPartsByMessageId: Map<string, Map<string, OpenCodeTextPartState>>;
342 turnTokenUsage: OpenCodeTurnTokenUsageAccumulator | undefined;352 turnTokenUsage: OpenCodeTurnTokenUsageAccumulator | undefined;
343 activeTurnId: TurnId | undefined;353 activeTurnId: TurnId | undefined;
344 activeAgent: string | undefined;354 activeAgent: string | undefined;
...@@ -374,7 +384,8 @@ interface OpenCodeTurnTokenUsageAccumulator {...@@ -374,7 +384,8 @@ interface OpenCodeTurnTokenUsageAccumulator {
374 readonly partIds: Set<string>;384 readonly partIds: Set<string>;
375 readonly promptMessageIds: Set<string>;385 readonly promptMessageIds: Set<string>;
376 readonly assistantOwnershipByMessageId: Map<string, "owned" | "other" | "unknown">;386 readonly assistantOwnershipByMessageId: Map<string, "owned" | "other" | "unknown">;
377 readonly unresolvedStepPartIds: Set<string>;387 // Native removal does not undo usage. Keep unresolved counts until this turn settles.
388 readonly unresolvedStepsByMessageId: Map<string, Map<string, OpenCodeStepUsage>>;
378 inputTokens: number;389 inputTokens: number;
379 cachedInputTokens: number;390 cachedInputTokens: number;
380 cacheCreationTokens: number;391 cacheCreationTokens: number;
...@@ -389,7 +400,7 @@ function makeOpenCodeTurnTokenUsageAccumulator(): OpenCodeTurnTokenUsageAccumula...@@ -389,7 +400,7 @@ function makeOpenCodeTurnTokenUsageAccumulator(): OpenCodeTurnTokenUsageAccumula
389 partIds: new Set(),400 partIds: new Set(),
390 promptMessageIds: new Set(),401 promptMessageIds: new Set(),
391 assistantOwnershipByMessageId: new Map(),402 assistantOwnershipByMessageId: new Map(),
392 unresolvedStepPartIds: new Set(),403 unresolvedStepsByMessageId: new Map(),
393 inputTokens: 0,404 inputTokens: 0,
394 cachedInputTokens: 0,405 cachedInputTokens: 0,
395 cacheCreationTokens: 0,406 cacheCreationTokens: 0,
...@@ -402,7 +413,7 @@ function makeOpenCodeTurnTokenUsageAccumulator(): OpenCodeTurnTokenUsageAccumula...@@ -402,7 +413,7 @@ function makeOpenCodeTurnTokenUsageAccumulator(): OpenCodeTurnTokenUsageAccumula
402413
403function accumulateOpenCodeStepUsage(414function accumulateOpenCodeStepUsage(
404 accumulator: OpenCodeTurnTokenUsageAccumulator,415 accumulator: OpenCodeTurnTokenUsageAccumulator,
405 part: Extract<Part, { readonly type: "step-finish" }>,416 part: OpenCodeStepUsage,
406): void {417): void {
407 if (accumulator.partIds.has(part.id)) return;418 if (accumulator.partIds.has(part.id)) return;
408 accumulator.partIds.add(part.id);419 accumulator.partIds.add(part.id);
...@@ -428,7 +439,9 @@ function takeOpenCodeTurnTokenUsage(...@@ -428,7 +439,9 @@ function takeOpenCodeTurnTokenUsage(
428 }439 }
429 return {440 return {
430 usageStatus:441 usageStatus:
431 complete && usage.complete && usage.unresolvedStepPartIds.size === 0 ? "complete" : "partial",442 complete && usage.complete && usage.unresolvedStepsByMessageId.size === 0
443 ? "complete"
444 : "partial",
432 usageScope: "main_agent",445 usageScope: "main_agent",
433 inputTokens: usage.inputTokens,446 inputTokens: usage.inputTokens,
434 cachedInputTokens: usage.cachedInputTokens,447 cachedInputTokens: usage.cachedInputTokens,
...@@ -579,18 +592,29 @@ function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray<UserI...@@ -579,18 +592,29 @@ function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray<UserI
579 }));592 }));
580}593}
581594
582function resolveTextStreamKind(part: Part | undefined): "assistant_text" | "reasoning_text" {595function resolveTextStreamKind(part: Pick<Part, "type">): "assistant_text" | "reasoning_text" {
583 return part?.type === "reasoning" ? "reasoning_text" : "assistant_text";596 return part.type === "reasoning" ? "reasoning_text" : "assistant_text";
584}597}
585598
586function textFromPart(part: Part): string | undefined {599function retainOpenCodeTextPart(
587 switch (part.type) {600 context: OpenCodeSessionContext,
588 case "text":601 part: OpenCodeTextPart,
589 case "reasoning":602): OpenCodeTextPartState {
590 return part.text;603 const parts =
591 default:604 context.textPartsByMessageId.get(part.messageID) ?? new Map<string, OpenCodeTextPartState>();
592 return undefined;605 const previous = parts.get(part.id);
593 }606 const state = {
607 id: part.id,
608 messageID: part.messageID,
609 type: part.type,
610 text: part.text,
611 ...(part.time !== undefined ? { time: part.time } : {}),
612 emittedText: previous?.emittedText,
613 completed: previous?.completed ?? false,
614 };
615 parts.set(part.id, state);
616 context.textPartsByMessageId.set(part.messageID, parts);
617 return state;
594}618}
595619
596function commonPrefixLength(left: string, right: string): number {620function commonPrefixLength(left: string, right: string): number {
...@@ -1362,6 +1386,7 @@ export function makeOpenCodeAdapter(...@@ -1362,6 +1386,7 @@ export function makeOpenCodeAdapter(
1362 if (message?.info.id === promptAdmission.messageId && message.info.role === "user") {1386 if (message?.info.id === promptAdmission.messageId && message.info.role === "user") {
1363 promptAdmission.messageObserved = true;1387 promptAdmission.messageObserved = true;
1364 context.messageRoleById.set(promptAdmission.messageId, "user");1388 context.messageRoleById.set(promptAdmission.messageId, "user");
1389 context.textPartsByMessageId.delete(promptAdmission.messageId);
1365 }1390 }
1366 }1391 }
13671392
...@@ -1568,35 +1593,23 @@ export function makeOpenCodeAdapter(...@@ -1568,35 +1593,23 @@ export function makeOpenCodeAdapter(
1568 /** Emit content.delta and item.completed events for an assistant text part. */1593 /** Emit content.delta and item.completed events for an assistant text part. */
1569 const emitAssistantTextDelta = Effect.fn("emitAssistantTextDelta")(function* (1594 const emitAssistantTextDelta = Effect.fn("emitAssistantTextDelta")(function* (
1570 context: OpenCodeSessionContext,1595 context: OpenCodeSessionContext,
1571 part: Part,1596 part: OpenCodeTextPartState,
1572 turnId: TurnId | undefined,1597 turnId: TurnId | undefined,
1573 raw: unknown,1598 raw: unknown,
1574 ) {1599 ) {
1575 const text = textFromPart(part);1600 if (part.text === undefined) {
1576 if (text === undefined) {
1577 return;1601 return;
1578 }1602 }
1579 const previousText = context.emittedTextByPartId.get(part.id);1603 const { latestText, deltaToEmit } = mergeOpenCodeAssistantText(part.emittedText, part.text);
1580 const { latestText, deltaToEmit } = mergeOpenCodeAssistantText(previousText, text);1604 part.emittedText = latestText;
1581 context.emittedTextByPartId.set(part.id, latestText);1605 part.text = latestText;
1582 if (latestText !== text) {
1583 context.partById.set(
1584 part.id,
1585 (part.type === "text" || part.type === "reasoning"
1586 ? { ...part, text: latestText }
1587 : part) satisfies Part,
1588 );
1589 }
1590 if (deltaToEmit.length > 0) {1606 if (deltaToEmit.length > 0) {
1591 yield* emit({1607 yield* emit({
1592 ...(yield* buildEventBase({1608 ...(yield* buildEventBase({
1593 threadId: context.session.threadId,1609 threadId: context.session.threadId,
1594 turnId,1610 turnId,
1595 itemId: part.id,1611 itemId: part.id,
1596 createdAt:1612 createdAt: part.time !== undefined ? isoFromEpochMs(part.time.start) : undefined,
1597 (part.type === "text" || part.type === "reasoning") && part.time !== undefined
1598 ? isoFromEpochMs(part.time.start)
1599 : undefined,
1600 raw,1613 raw,
1601 })),1614 })),
1602 type: "content.delta",1615 type: "content.delta",
...@@ -1607,12 +1620,8 @@ export function makeOpenCodeAdapter(...@@ -1607,12 +1620,8 @@ export function makeOpenCodeAdapter(
1607 });1620 });
1608 }1621 }
16091622
1610 if (1623 if (part.type === "text" && part.time?.end !== undefined && !part.completed) {
1611 part.type === "text" &&1624 part.completed = true;
1612 part.time?.end !== undefined &&
1613 !context.completedAssistantPartIds.has(part.id)
1614 ) {
1615 context.completedAssistantPartIds.add(part.id);
1616 yield* emit({1625 yield* emit({
1617 ...(yield* buildEventBase({1626 ...(yield* buildEventBase({
1618 threadId: context.session.threadId,1627 threadId: context.session.threadId,
...@@ -2315,6 +2324,9 @@ export function makeOpenCodeAdapter(...@@ -2315,6 +2324,9 @@ export function makeOpenCodeAdapter(
2315 }2324 }
2316 }2325 }
2317 context.messageRoleById.set(event.properties.info.id, event.properties.info.role);2326 context.messageRoleById.set(event.properties.info.id, event.properties.info.role);
2327 if (event.properties.info.role === "user") {
2328 context.textPartsByMessageId.delete(event.properties.info.id);
2329 }
2318 if (event.properties.info.role === "assistant") {2330 if (event.properties.info.role === "assistant") {
2319 const usage = context.turnTokenUsage;2331 const usage = context.turnTokenUsage;
2320 const parentMessageId =2332 const parentMessageId =
...@@ -2337,15 +2349,19 @@ export function makeOpenCodeAdapter(...@@ -2337,15 +2349,19 @@ export function makeOpenCodeAdapter(
2337 : priorOwnership;2349 : priorOwnership;
2338 if (usage) {2350 if (usage) {
2339 usage.assistantOwnershipByMessageId.set(event.properties.info.id, ownership);2351 usage.assistantOwnershipByMessageId.set(event.properties.info.id, ownership);
2340 }2352 if (ownership !== "unknown") {
2341 for (const part of context.partById.values()) {2353 const steps = usage.unresolvedStepsByMessageId.get(event.properties.info.id);
2342 if (part.messageID !== event.properties.info.id) {2354 if (ownership === "owned" && steps) {
2343 continue;2355 for (const step of steps.values()) {
2344 }2356 accumulateOpenCodeStepUsage(usage, step);
2345 if (usage && part.type === "step-finish") {2357 }
2346 if (ownership !== "unknown") usage.unresolvedStepPartIds.delete(part.id);2358 }
2347 if (ownership === "owned") accumulateOpenCodeStepUsage(usage, part);2359 usage.unresolvedStepsByMessageId.delete(event.properties.info.id);
2348 }2360 }
2361 }
2362 for (const part of context.textPartsByMessageId
2363 .get(event.properties.info.id)
2364 ?.values() ?? []) {
2349 yield* emitAssistantTextDelta(context, part, turnId, event);2365 yield* emitAssistantTextDelta(context, part, turnId, event);
2350 }2366 }
2351 }2367 }
...@@ -2354,16 +2370,24 @@ export function makeOpenCodeAdapter(...@@ -2354,16 +2370,24 @@ export function makeOpenCodeAdapter(
23542370
2355 case "message.removed": {2371 case "message.removed": {
2356 context.messageRoleById.delete(event.properties.messageID);2372 context.messageRoleById.delete(event.properties.messageID);
2373 context.textPartsByMessageId.delete(event.properties.messageID);
2374 break;
2375 }
2376
2377 case "message.part.removed": {
2378 const parts = context.textPartsByMessageId.get(event.properties.messageID);
2379 parts?.delete(event.properties.partID);
2380 if (parts?.size === 0) {
2381 context.textPartsByMessageId.delete(event.properties.messageID);
2382 }
2357 break;2383 break;
2358 }2384 }
23592385
2360 case "message.part.delta": {2386 case "message.part.delta": {
2361 const existingPart = context.partById.get(event.properties.partID);2387 const existingPart = context.textPartsByMessageId
2362 if (2388 .get(event.properties.messageID)
2363 !existingPart ||2389 ?.get(event.properties.partID);
2364 (existingPart.type !== "text" && existingPart.type !== "reasoning") ||2390 if (existingPart?.text === undefined || event.properties.field !== "text") {
2365 event.properties.field !== "text"
2366 ) {
2367 break;2391 break;
2368 }2392 }
2369 const role = messageRoleForPart(context, existingPart);2393 const role = messageRoleForPart(context, existingPart);
...@@ -2375,21 +2399,13 @@ export function makeOpenCodeAdapter(...@@ -2375,21 +2399,13 @@ export function makeOpenCodeAdapter(
2375 if (delta.length === 0) {2399 if (delta.length === 0) {
2376 break;2400 break;
2377 }2401 }
2378 const previousText =2402 const previousText = existingPart.emittedText ?? existingPart.text;
2379 context.emittedTextByPartId.get(event.properties.partID) ??
2380 textFromPart(existingPart) ??
2381 "";
2382 const { nextText, deltaToEmit } = appendOpenCodeAssistantTextDelta(previousText, delta);2403 const { nextText, deltaToEmit } = appendOpenCodeAssistantTextDelta(previousText, delta);
2383 if (deltaToEmit.length === 0) {2404 if (deltaToEmit.length === 0) {
2384 break;2405 break;
2385 }2406 }
2386 context.emittedTextByPartId.set(event.properties.partID, nextText);2407 existingPart.emittedText = nextText;
2387 if (existingPart.type === "text" || existingPart.type === "reasoning") {2408 existingPart.text = nextText;
2388 context.partById.set(event.properties.partID, {
2389 ...existingPart,
2390 text: nextText,
2391 });
2392 }
2393 yield* emit({2409 yield* emit({
2394 ...(yield* buildEventBase({2410 ...(yield* buildEventBase({
2395 threadId: context.session.threadId,2411 threadId: context.session.threadId,
...@@ -2408,29 +2424,38 @@ export function makeOpenCodeAdapter(...@@ -2408,29 +2424,38 @@ export function makeOpenCodeAdapter(
24082424
2409 case "message.part.updated": {2425 case "message.part.updated": {
2410 const part = event.properties.part;2426 const part = event.properties.part;
2411 // Tool events use the incoming part and do not need a cached copy.
2412 if (part.type !== "tool") {
2413 context.partById.set(part.id, part);
2414 }
2415 const messageRole = messageRoleForPart(context, part);2427 const messageRole = messageRoleForPart(context, part);
24162428
2417 if (turnId && part.type === "step-finish" && context.turnTokenUsage) {2429 if (turnId && part.type === "step-finish" && context.turnTokenUsage) {
2418 const ownership = context.turnTokenUsage.assistantOwnershipByMessageId.get(2430 const usage = context.turnTokenUsage;
2419 part.messageID,2431 const ownership = usage.assistantOwnershipByMessageId.get(part.messageID);
2420 );
2421 if (ownership === "owned") {2432 if (ownership === "owned") {
2422 accumulateOpenCodeStepUsage(context.turnTokenUsage, part);2433 accumulateOpenCodeStepUsage(usage, part);
2423 } else if (2434 } else if (
2424 ownership === "unknown" ||2435 ownership === "unknown" ||
2425 (ownership === undefined &&2436 (ownership === undefined &&
2426 context.messageRoleById.get(part.messageID) !== "assistant")2437 context.messageRoleById.get(part.messageID) !== "assistant")
2427 ) {2438 ) {
2428 context.turnTokenUsage.unresolvedStepPartIds.add(part.id);2439 const steps =
2440 usage.unresolvedStepsByMessageId.get(part.messageID) ??
2441 new Map<string, OpenCodeStepUsage>();
2442 steps.set(part.id, { id: part.id, tokens: part.tokens });
2443 usage.unresolvedStepsByMessageId.set(part.messageID, steps);
2429 }2444 }
2430 }2445 }
24312446
2432 if (messageRole === "assistant") {2447 if ((part.type === "text" || part.type === "reasoning") && messageRole !== "user") {
2433 yield* emitAssistantTextDelta(context, part, turnId, event);2448 const state = retainOpenCodeTextPart(context, part);
2449 if (messageRole === "assistant") {
2450 yield* emitAssistantTextDelta(context, state, turnId, event);
2451 }
2452 } else {
2453 const previous = context.textPartsByMessageId.get(part.messageID)?.get(part.id);
2454 if (previous) {
2455 // A non-text PATCH removes the current snapshot. Keep emitted text
2456 // so a later text PATCH still emits only the changed suffix.
2457 previous.text = undefined;
2458 }
2434 }2459 }
24352460
2436 if (part.type === "tool") {2461 if (part.type === "tool") {
...@@ -2963,10 +2988,8 @@ export function makeOpenCodeAdapter(...@@ -2963,10 +2988,8 @@ export function makeOpenCodeAdapter(
2963 requestRelationRetries: new Map(),2988 requestRelationRetries: new Map(),
2964 pendingPermissions: new Map(),2989 pendingPermissions: new Map(),
2965 pendingQuestions: new Map(),2990 pendingQuestions: new Map(),
2966 partById: new Map(),2991 textPartsByMessageId: new Map(),
2967 emittedTextByPartId: new Map(),
2968 messageRoleById: new Map(),2992 messageRoleById: new Map(),
2969 completedAssistantPartIds: new Set(),
2970 turnTokenUsage: undefined,2993 turnTokenUsage: undefined,
2971 activeTurnId: undefined,2994 activeTurnId: undefined,
2972 activeAgent: undefined,2995 activeAgent: undefined,