jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "fix(server): wait for opencode completion after EOF"

Browse Source 
This reverts commit 2a84b081298f904936f5ad61dca2388b48835e0e.
This commit is contained in:
Mohamed Boudra
2026-05-10 18:28:53 +07:00
parent 3014576c4c
commit ef892bd27d
2 changed files with 6 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/opencode-agent.test.ts
-308
View File
@@ -13,7 +13,6 @@ import {
} from "./opencode-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
import type {
AgentSession,
AgentSessionConfig,
AgentStreamEvent,
ToolCallTimelineItem,
@@ -43,8 +42,6 @@ interface TurnResult {
error?: string;
}
const FAST_OPENCODE_EOF_RECOVERY_POLICY = { maxAttempts: 5, delayMs: 1 };
async function collectTurnEvents(iterator: AsyncGenerator<AgentStreamEvent>): Promise<TurnResult> {
const result: TurnResult = {
events: [],
@@ -81,140 +78,6 @@ async function collectTurnEvents(iterator: AsyncGenerator<AgentStreamEvent>): Pr
return result;
}
interface FakeOpenCodeStream {
stream: AsyncIterable<OpenCodeEvent>;
close: () => void;
closed: Promise<void>;
}
function createFakeOpenCodeStream(events: OpenCodeEvent[] = []): FakeOpenCodeStream {
let releaseStream!: () => void;
const canEnd = new Promise<void>((resolve) => {
releaseStream = resolve;
});
let notifyEnded!: () => void;
const closed = new Promise<void>((resolve) => {
notifyEnded = resolve;
});
return {
close: releaseStream,
closed,
stream: {
[Symbol.asyncIterator]: () => {
let index = 0;
return {
next: async () => {
if (index < events.length) {
return { done: false, value: events[index++] };
}
await canEnd;
notifyEnded();
return { done: true, value: undefined };
},
};
},
},
};
}
function createFakeOpenCode(stream: FakeOpenCodeStream, onPrompt: () => void | Promise<void>) {
return {
event: {
subscribe: async () => ({ stream: stream.stream }),
},
provider: {
list: async () => ({ data: { connected: [], all: [] }, error: undefined }),
},
session: {
create: async () => ({ data: { id: "ses_unit_test" }, error: undefined }),
promptAsync: async () => {
await onPrompt();
return { data: {}, error: undefined };
},
abort: async () => ({ data: true, error: undefined }),
update: async () => ({ data: true, error: undefined }),
delete: async () => ({ data: true, error: undefined }),
},
} as never;
}
async function createSessionWithFakeOpenCode(params: {
storageRoot: string;
cwd: string;
stream: FakeOpenCodeStream;
onPrompt: () => void | Promise<void>;
}): Promise<AgentSession> {
const fakeClient = createFakeOpenCode(params.stream, params.onPrompt);
const client = new OpenCodeAgentClient(createTestLogger(), undefined, params.storageRoot, {
runtime: {
acquireServer: async () => ({
server: { port: 0, url: "http://localhost" },
release: () => {},
}),
ensureServerRunning: async () => ({ port: 0, url: "http://localhost" }),
createClient: () => fakeClient,
shutdown: async () => undefined,
},
eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
});
return client.createSession({ provider: "opencode", cwd: params.cwd });
}
function observeTurn(session: AgentSession) {
const events: AgentStreamEvent[] = [];
let resolveTerminal!: (event: AgentStreamEvent) => void;
const terminal = new Promise<AgentStreamEvent>((resolve) => {
resolveTerminal = resolve;
});
session.subscribe((event) => {
events.push(event);
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
resolveTerminal(event);
}
});
return {
terminal,
assistantMessages: () =>
events
.flatMap((event) => (event.type === "timeline" ? [event.item] : []))
.filter((item): item is AssistantMessageTimelineItem => item.type === "assistant_message"),
};
}
function openCodeAssistantStarted(messageId: string): OpenCodeEvent {
return {
type: "message.updated",
properties: {
info: {
id: messageId,
sessionID: "ses_unit_test",
role: "assistant",
},
},
} as OpenCodeEvent;
}
function openCodeTextDelta(messageId: string, partId: string, delta: string): OpenCodeEvent {
return {
type: "message.part.delta",
properties: {
sessionID: "ses_unit_test",
messageID: messageId,
partID: partId,
field: "text",
delta,
},
} as OpenCodeEvent;
}
function isBinaryInstalled(binary: string): boolean {
try {
const out = execFileSync("which", [binary], { encoding: "utf8" }).trim();
@@ -808,7 +671,6 @@ describe("OpenCode adapter startTurn error handling", () => {
createClient: vi.fn().mockReturnValue(fakeClient),
shutdown: vi.fn().mockResolvedValue(undefined),
},
eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
});
const session = await client.createSession({ provider: "opencode", cwd });
@@ -834,126 +696,6 @@ describe("OpenCode adapter startTurn error handling", () => {
rmSync(storageRoot, { recursive: true, force: true });
});
test("recovers SSE EOF when persisted assistant completion appears after the stream closes", async () => {
const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
const cwd = "/tmp/test";
const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
try {
writeOpenCodeStoredSession(storageRoot, cwd);
const stream = createFakeOpenCodeStream();
const session = await createSessionWithFakeOpenCode({
storageRoot,
cwd,
stream,
onPrompt: stream.close,
});
const turn = observeTurn(session);
await session.startTurn("hello");
await stream.closed;
writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply");
expect((await turn.terminal).type).toBe("turn_completed");
expect(
turn
.assistantMessages()
.map((message) => message.text)
.join(""),
).toBe("Recovered assistant reply");
} finally {
dateNowSpy.mockRestore();
rmSync(storageRoot, { recursive: true, force: true });
}
});
test("recovers delayed persisted completion without duplicating text already streamed before EOF", async () => {
const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
const cwd = "/tmp/test";
const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
let completionWriteSettled: Promise<void> = Promise.resolve();
const writeDelayedCompletion = async () => {
await nextMacrotask();
writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply");
};
try {
writeOpenCodeStoredSession(storageRoot, cwd);
const stream = createFakeOpenCodeStream([
openCodeAssistantStarted("msg_assistant"),
openCodeTextDelta("msg_assistant", "prt_text", "Recovered "),
]);
const session = await createSessionWithFakeOpenCode({
storageRoot,
cwd,
stream,
onPrompt: async () => {
stream.close();
completionWriteSettled = writeDelayedCompletion();
await completionWriteSettled;
},
});
const turn = observeTurn(session);
await session.startTurn("hello");
await stream.closed;
expect((await turn.terminal).type).toBe("turn_completed");
const assistantMessages = turn.assistantMessages();
expect(assistantMessages.map((message) => message.text).join("")).toBe(
"Recovered assistant reply",
);
expect(assistantMessages.map((message) => message.text)).toEqual([
"Recovered ",
"assistant reply",
]);
} finally {
await completionWriteSettled;
dateNowSpy.mockRestore();
rmSync(storageRoot, { recursive: true, force: true });
}
});
test("ignores old completed assistant messages while waiting for delayed current-turn completion", async () => {
const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
const cwd = "/tmp/test";
const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
try {
writeOpenCodeStoredSession(storageRoot, cwd);
writeOpenCodeAssistantCompletion(storageRoot, "msg_assistant_old", "Old assistant reply", {
created: 1500,
completed: 1600,
partStart: 1500,
partEnd: 1550,
});
const stream = createFakeOpenCodeStream();
const session = await createSessionWithFakeOpenCode({
storageRoot,
cwd,
stream,
onPrompt: stream.close,
});
const turn = observeTurn(session);
await session.startTurn("hello");
await stream.closed;
writeRecoveredAssistantCompletion(
storageRoot,
"msg_assistant_current",
"Current assistant reply",
);
expect((await turn.terminal).type).toBe("turn_completed");
expect(turn.assistantMessages().map((message) => message.text)).toEqual([
"Current assistant reply",
]);
} finally {
dateNowSpy.mockRestore();
rmSync(storageRoot, { recursive: true, force: true });
}
});
test("keeps SSE EOF as turn_failed without persisted completion evidence", async () => {
const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
const cwd = "/tmp/test";
@@ -1021,7 +763,6 @@ describe("OpenCode adapter startTurn error handling", () => {
createClient: vi.fn().mockReturnValue(fakeClient),
shutdown: vi.fn().mockResolvedValue(undefined),
},
eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
});
const session = await client.createSession({ provider: "opencode", cwd });
@@ -1126,7 +867,6 @@ describe("OpenCode adapter startTurn error handling", () => {
createClient: vi.fn().mockReturnValue(fakeClient),
shutdown: vi.fn().mockResolvedValue(undefined),
},
eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
});
const session = await client.createSession({ provider: "opencode", cwd });
@@ -1318,51 +1058,3 @@ function writeOpenCodeJson(storageRoot: string, relativePath: string, value: unk
mkdirSync(path.dirname(filePath), { recursive: true });
writeFileSync(filePath, JSON.stringify(value), "utf8");
}
function writeOpenCodeStoredSession(storageRoot: string, cwd: string): void {
writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", {
id: "ses_unit_test",
directory: cwd,
time: { created: 1000, updated: 3000 },
});
}
function nextMacrotask(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0));
}
function writeRecoveredAssistantCompletion(
storageRoot: string,
messageId: string,
text: string,
): void {
writeOpenCodeAssistantCompletion(storageRoot, messageId, text, {
created: 2000,
completed: 2500,
partStart: 2100,
partEnd: 2400,
});
}
function writeOpenCodeAssistantCompletion(
storageRoot: string,
messageId: string,
text: string,
time: { created: number; completed: number; partStart: number; partEnd: number },
): void {
writeOpenCodeJson(storageRoot, `message/ses_unit_test/${messageId}.json`, {
id: messageId,
sessionID: "ses_unit_test",
role: "assistant",
finish: "stop",
time: { created: time.created, completed: time.completed },
});
writeOpenCodeJson(storageRoot, `part/${messageId}/prt_text.json`, {
id: "prt_text",
sessionID: "ses_unit_test",
messageID: messageId,
type: "text",
text,
time: { start: time.partStart, end: time.partEnd },
});
}
packages/server/src/server/agent/providers/opencode-agent.ts
+6 -49
View File
@@ -74,10 +74,6 @@ const OPENCODE_CAPABILITIES: AgentCapabilityFlags = {
const OPENCODE_BUILD_MODE_ID = "build";
const OPENCODE_FULL_ACCESS_MODE_ID = "full-access";
const OPENCODE_STORAGE_SESSION_LIMIT = 200;
const DEFAULT_OPENCODE_EOF_RECOVERY_POLICY = {
maxAttempts: 600,
delayMs: 1_000,
};
const DEFAULT_MODES: AgentMode[] = [
{
@@ -148,11 +144,6 @@ interface OpenCodePersistedAssistantCompletion {
usage?: AgentUsage;
}
interface OpenCodeEofRecoveryPolicy {
maxAttempts: number;
delayMs: number;
}
type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" };
type OpenCodeMessageRole = "user" | "assistant";
@@ -913,10 +904,6 @@ async function readOpenCodePersistedAssistantCompletion(
return null;
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function hasStrongPersistedCompletionEvidence(
message: OpenCodeStoredMessage,
parts: OpenCodeStoredPart[],
@@ -1056,7 +1043,6 @@ export const __openCodeInternals = {
interface OpenCodeAgentClientDeps {
runtime?: OpenCodeRuntime;
eofRecoveryPolicy?: OpenCodeEofRecoveryPolicy;
}
class ProductionOpenCodeRuntime implements OpenCodeRuntime {
@@ -1088,7 +1074,6 @@ export class OpenCodeAgentClient implements AgentClient {
private readonly runtimeSettings?: ProviderRuntimeSettings;
private readonly modelContextWindows = new Map<string, number>();
private readonly storageRoot: string;
private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy;
constructor(
logger: Logger,
@@ -1099,7 +1084,6 @@ export class OpenCodeAgentClient implements AgentClient {
this.logger = logger.child({ module: "agent", provider: "opencode" });
this.runtimeSettings = runtimeSettings;
this.storageRoot = storageRoot ?? resolveOpenCodeStorageRoot();
this.eofRecoveryPolicy = deps.eofRecoveryPolicy ?? DEFAULT_OPENCODE_EOF_RECOVERY_POLICY;
this.runtime =
deps.runtime ??
new ProductionOpenCodeRuntime(
@@ -1147,7 +1131,6 @@ export class OpenCodeAgentClient implements AgentClient {
new Map(this.modelContextWindows),
acquisition.release,
options?.persistSession,
this.eofRecoveryPolicy,
);
} catch (error) {
acquisition.release();
@@ -1189,8 +1172,6 @@ export class OpenCodeAgentClient implements AgentClient {
this.storageRoot,
new Map(this.modelContextWindows),
acquisition.release,
undefined,
this.eofRecoveryPolicy,
);
} catch (error) {
acquisition.release();
@@ -2321,7 +2302,6 @@ class OpenCodeAgentSession implements AgentSession {
private selectedModelContextWindowMaxTokens: number | undefined;
private releaseServer: (() => void) | null;
private readonly persistSession: boolean;
private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy;
private deletedFromProvider = false;
private foregroundAssistantMessageEmitted = false;
private foregroundAssistantText = "";
@@ -2337,7 +2317,6 @@ class OpenCodeAgentSession implements AgentSession {
modelContextWindowsByModelKey: ReadonlyMap<string, number> = new Map(),
releaseServer?: () => void,
persistSession = true,
eofRecoveryPolicy: OpenCodeEofRecoveryPolicy = DEFAULT_OPENCODE_EOF_RECOVERY_POLICY,
) {
this.config = config;
this.client = client;
@@ -2348,7 +2327,6 @@ class OpenCodeAgentSession implements AgentSession {
this.currentMode = normalizeOpenCodeModeId(config.modeId);
this.releaseServer = releaseServer ?? null;
this.persistSession = persistSession;
this.eofRecoveryPolicy = eofRecoveryPolicy;
this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens(
config.model,
);
@@ -2692,7 +2670,12 @@ class OpenCodeAgentSession implements AgentSession {
return false;
}
const completion = await this.waitForPersistedAssistantCompletion(turnId);
const completion = await readOpenCodePersistedAssistantCompletion(
this.storageRoot,
this.sessionId,
this.foregroundKnownMessageIds,
this.foregroundTurnStartedAt,
);
if (!completion || this.activeForegroundTurnId !== turnId) {
return false;
}
@@ -2749,32 +2732,6 @@ class OpenCodeAgentSession implements AgentSession {
return true;
}
private async waitForPersistedAssistantCompletion(
turnId: string,
): Promise<OpenCodePersistedAssistantCompletion | null> {
for (let attempt = 0; attempt < this.eofRecoveryPolicy.maxAttempts; attempt += 1) {
if (this.foregroundTurnStartedAt === null || this.activeForegroundTurnId !== turnId) {
return null;
}
const completion = await readOpenCodePersistedAssistantCompletion(
this.storageRoot,
this.sessionId,
this.foregroundKnownMessageIds,
this.foregroundTurnStartedAt,
);
if (completion) {
return completion;
}
if (attempt < this.eofRecoveryPolicy.maxAttempts - 1) {
await sleep(this.eofRecoveryPolicy.delayMs);
}
}
return null;
}
private resolvePersistedAssistantRecoveryText(completedText: string): string | null {
if (!this.foregroundAssistantMessageEmitted) {
return completedText;