diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index ecc1852b6..b13fe0768 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -1996,7 +1996,7 @@ describe("OpenCode adapter startTurn error handling", () => { } }); - test("streams an autonomous wake as soon as the canceled run reaches its terminal", async () => { + test("streams an autonomous wake after the canceled run reaches its terminal", async () => { const { parent: session, openCode } = await createParentSession("ses_wake_after_terminal"); const settleAbort = createTestDeferred(); const events: AgentStreamEvent[] = []; @@ -2023,6 +2023,13 @@ describe("OpenCode adapter startTurn error handling", () => { type: "session.idle", properties: { sessionID: "ses_wake_after_terminal" }, }); + openCode.emitEvent({ + type: "session.status", + properties: { + sessionID: "ses_wake_after_terminal", + status: { type: "busy" }, + }, + }); for (const event of userMessageEvents({ sessionId: "ses_wake_after_terminal", messageId: "msg_wake_after_terminal", @@ -2137,6 +2144,10 @@ describe("OpenCode adapter startTurn error handling", () => { }, }, }); + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_deferred_autonomous", status: { type: "busy" } }, + }); openCode.emitEvent({ type: "message.part.updated", properties: { @@ -2372,6 +2383,13 @@ describe("OpenCode adapter startTurn error handling", () => { type: "session.idle", properties: { sessionID: "ses_carried_abort" }, }); + openCode.emitEvent({ + type: "session.status", + properties: { + sessionID: "ses_carried_abort", + status: { type: "busy" }, + }, + }); for (const event of userMessageEvents({ sessionId: "ses_carried_abort", messageId: "msg_wake_between_stops", @@ -3082,7 +3100,9 @@ describe("OpenCode persisted sessions", () => { }); describe("OpenCode provider subagent contract", () => { - async function createAdoptedChildSession(): Promise<{ + async function createAdoptedChildSession( + configureChild?: (client: TestOpenCodeClient) => void, + ): Promise<{ readonly runtime: TestOpenCodeHarness; readonly provider: OpenCodeAgentClient; readonly parent: Awaited>; @@ -3092,6 +3112,7 @@ describe("OpenCode provider subagent contract", () => { const runtime = new TestOpenCodeHarness(); const parentClient = new TestOpenCodeClient(); const childClient = new TestOpenCodeClient(); + configureChild?.(childClient); parentClient.sessionCreateResponse = { data: { id: "ses_parent_external" } }; runtime.enqueueClient(parentClient); runtime.enqueueClient(childClient); @@ -3233,14 +3254,14 @@ describe("OpenCode provider subagent contract", () => { }); for (const event of [ - ...assistantTurnEvents({ sessionId: "ses_child_external", text: "child says hi" }).slice( - 0, - 2, - ), { type: "session.status", properties: { sessionID: "ses_child_external", status: { type: "busy" } }, }, + ...assistantTurnEvents({ sessionId: "ses_child_external", text: "child says hi" }).slice( + 0, + 2, + ), { type: "session.idle", properties: { sessionID: "ses_child_external" } }, ]) { childClient.emitEvent(event); @@ -3272,7 +3293,108 @@ describe("OpenCode provider subagent contract", () => { ); }); - test("surfaces an autonomous OpenCode wake as a new parent turn", async () => { + test.each(["busy", "retry"] as const)( + "reconciles an adopted child that was already %s before attachment", + async (runnerStatus) => { + const { child, childClient, parent } = await createAdoptedChildSession((client) => { + client.sessionStatusResponse = { + data: { ses_child_external: { type: runnerStatus } }, + }; + }); + const completed = createTestDeferred(); + const events: AgentStreamEvent[] = []; + try { + child.subscribe((event) => { + events.push(event); + if (event.type === "turn_completed") { + completed.resolve(); + } + }); + + await vi.waitFor(() => { + expect(events).toContainEqual({ + type: "turn_started", + provider: "opencode", + turnId: "opencode-turn-0", + }); + }); + for (const event of assistantTurnEvents({ + sessionId: "ses_child_external", + text: "already-running child says hi", + })) { + childClient.emitEvent(event); + } + + await completed.promise; + + expect(childClient.calls.sessionStatus).toEqual([{ directory: "/workspace/repo" }]); + expect(turnEventSignatures(events)).toEqual([ + ["turn_started", "opencode-turn-0"], + ["timeline", "opencode-turn-0"], + ["turn_completed", "opencode-turn-0"], + ]); + expect(events).toContainEqual( + expect.objectContaining({ + type: "timeline", + item: expect.objectContaining({ text: "already-running child says hi" }), + }), + ); + } finally { + await child.close(); + await parent.close(); + } + }, + ); + + test("ignores a stale busy snapshot after the live stream reaches idle", async () => { + const statusStarted = createTestDeferred(); + const releaseStatus = createTestDeferred(); + const { child, childClient, parent } = await createAdoptedChildSession((client) => { + client.sessionStatusImplementation = async () => { + statusStarted.resolve(); + await releaseStatus.promise; + return { data: { ses_child_external: { type: "busy" } } }; + }; + }); + const idleConsumed = createTestDeferred(); + const events: AgentStreamEvent[] = []; + try { + child.subscribe((event) => { + events.push(event); + if (event.type === "provider_subagent") { + idleConsumed.resolve(); + } + }); + + await statusStarted.promise; + childClient.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_child_external" }, + }); + childClient.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_child_idle_drain_marker", + parentID: "ses_child_external", + title: "Idle drain marker", + }, + }, + }); + await idleConsumed.promise; + + releaseStatus.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(events.some((event) => event.type === "turn_started")).toBe(false); + } finally { + releaseStatus.resolve(); + await child.close(); + await parent.close(); + } + }); + + test("uses busy as the autonomous boundary without replaying earlier idle content", async () => { const { parent, openCode } = await createParentSession("ses_parent_plugin_wake"); const events: AgentStreamEvent[] = []; parent.subscribe((event) => events.push(event)); @@ -3285,6 +3407,22 @@ describe("OpenCode provider subagent contract", () => { })) { openCode.emitEvent(event); } + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_plugin_wake_drain_marker", + parentID: "ses_parent_plugin_wake", + title: "Pre-busy drain marker", + }, + }, + }); + await vi.waitFor(() => { + expect(events).toContainEqual(expect.objectContaining({ type: "provider_subagent" })); + }); + expect(events.filter((event) => event.type !== "provider_subagent")).toEqual([]); + events.length = 0; + openCode.emitEvent({ type: "session.status", properties: { sessionID: "ses_parent_plugin_wake", status: { type: "busy" } }, @@ -3300,7 +3438,6 @@ describe("OpenCode provider subagent contract", () => { expect(turnEventSignatures(events)).toEqual([ ["turn_started", "opencode-turn-0"], ["timeline", "opencode-turn-0"], - ["timeline", "opencode-turn-0"], ["turn_completed", "opencode-turn-0"], ]); }); @@ -3313,6 +3450,43 @@ describe("OpenCode provider subagent contract", () => { }), }), ); + expect(events).not.toContainEqual( + expect.objectContaining({ + type: "timeline", + item: expect.objectContaining({ + text: "All background tasks are complete.", + }), + }), + ); + } finally { + await parent.close(); + } + }); + + test("starts autonomous activity from busy without requiring an initiating message", async () => { + const { parent, openCode } = await createParentSession("ses_parent_busy_only"); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + + try { + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_parent_busy_only", status: { type: "busy" } }, + }); + for (const event of assistantTurnEvents({ + sessionId: "ses_parent_busy_only", + text: "Autonomous response after busy.", + })) { + openCode.emitEvent(event); + } + + await vi.waitFor(() => { + expect(turnEventSignatures(events)).toEqual([ + ["turn_started", "opencode-turn-0"], + ["timeline", "opencode-turn-0"], + ["turn_completed", "opencode-turn-0"], + ]); + }); } finally { await parent.close(); } @@ -3378,6 +3552,10 @@ describe("OpenCode provider subagent contract", () => { text: "Autonomous wake", })[0], ); + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_parent_active_wake", status: { type: "busy" } }, + }); await vi.waitFor(() => { expect(events).toEqual([ { type: "turn_started", provider: "opencode", turnId: "opencode-turn-0" }, @@ -3519,6 +3697,10 @@ describe("OpenCode provider subagent contract", () => { } }); + childClient.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_child_external", status: { type: "busy" } }, + }); childClient.emitEvent({ type: "permission.asked", properties: { @@ -4446,3 +4628,53 @@ function waitForAbort(signal: AbortSignal): Promise { signal.addEventListener("abort", () => resolve(), { once: true }); }); } + +describe("OpenCode snapshot summary false-idle regression", () => { + test("user message.updated carrying snapshot diffs does not start a turn", async () => { + const runtime = new TestOpenCodeHarness(); + const openCodeClient = new TestOpenCodeClient(); + openCodeClient.sessionCreateResponse = { data: { id: "ses_snapshot" } }; + runtime.enqueueClient(openCodeClient); + const client = new OpenCodeAgentClient(createTestLogger(), undefined, { + serverManager: runtime, + createClient: runtime.createClient, + }); + const session = await client.createSession( + { provider: "opencode", cwd: "/workspace/repo" }, + { env: { PASEO_AGENT_ID: "snapshot-agent" } }, + ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + // Captured live shape: after the turn goes idle OpenCode writes the filesystem + // snapshot diff onto the user message, with no time.end on the message. + openCodeClient.emitEvent({ + type: "message.updated", + properties: { + info: { + id: "msg_snapshot_user", + sessionID: "ses_snapshot", + role: "user", + time: { created: 1000 }, + summary: { + diffs: [{ file: "packages/server/src/thing.ts", additions: 3, deletions: 1 }], + }, + }, + }, + }); + // Drain marker: events are processed in order, so observing this one proves + // the snapshot update above was already handled. + openCodeClient.emitEvent({ + type: "session.created", + properties: { + info: { id: "ses_snapshot_child", parentID: "ses_snapshot", title: "Drain marker" }, + }, + }); + await vi.waitFor(() => { + expect(events).toContainEqual(expect.objectContaining({ type: "provider_subagent" })); + }); + + expect(events.some((event) => event.type === "turn_started")).toBe(false); + await session.close(); + }); +}); diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 99c542bea..672c510dc 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -2815,15 +2815,28 @@ function getOpenCodeEventSessionId(event: OpenCodeEvent): string | null { ); } -function isOpenCodeTerminalEvent(event: OpenCodeEvent, sessionId: string): boolean { - if (event.type === "session.idle" || event.type === "session.error") { - return event.properties.sessionID === sessionId; +function getOpenCodeRunnerStatusFromEvent( + event: OpenCodeEvent, + sessionId: string, +): OpenCodeRunnerStatus | null { + if (getOpenCodeEventSessionId(event) !== sessionId) { + return null; } - return ( - event.type === "session.status" && - event.properties.sessionID === sessionId && - event.properties.status.type === "idle" - ); + if (event.type === "session.status") { + return event.properties.status.type; + } + if (event.type === "session.idle" || event.type === "session.error") { + return "idle"; + } + return null; +} + +function isOpenCodeRunnerActive(status: OpenCodeRunnerStatus): boolean { + return status === "busy" || status === "retry"; +} + +function isOpenCodeTerminalEvent(event: OpenCodeEvent, sessionId: string): boolean { + return getOpenCodeRunnerStatusFromEvent(event, sessionId) === "idle"; } function isOpenCodeProviderInternalEvent(event: AgentStreamEvent): boolean { @@ -2937,6 +2950,8 @@ class OpenCodeAgentSession implements AgentSession { * run, and a rejection means we never proved the runner stopped. */ private abortSettlement: Promise = Promise.resolve(); + private externalStatusReconciliationStarted = false; + private runnerStatusRevision = 0; private readonly runningToolCalls = new Map(); private subAgentsByCallId = new Map(); private subAgentCallIdByChildSessionId = new Map(); @@ -3393,12 +3408,40 @@ class OpenCodeAgentSession implements AgentSession { } subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); + this.startExternalStatusReconciliation(); this.startChildSessionHydration(); return () => { this.subscribers.delete(callback); }; } + private startExternalStatusReconciliation(): void { + if (!this.externallyDriven || this.externalStatusReconciliationStarted || this.closed) { + return; + } + this.externalStatusReconciliationStarted = true; + void this.reconcileExternalRunnerStatus().catch((error) => { + this.logger.warn( + { err: error, sessionId: this.sessionId }, + "Failed to reconcile externally driven OpenCode session status", + ); + }); + } + + private async reconcileExternalRunnerStatus(): Promise { + await this.ensureEventStreamReady(); + const observedRevision = this.runnerStatusRevision; + const runnerStatus = await this.readProviderRunnerStatus(); + if ( + this.runnerStatusRevision !== observedRevision || + this.turnState.status !== "idle" || + !isOpenCodeRunnerActive(runnerStatus) + ) { + return; + } + this.startAutonomousTurn(); + } + private startChildSessionHydration(): void { if (this.childHydrationPromise) { return; @@ -3657,21 +3700,8 @@ class OpenCodeAgentSession implements AgentSession { if (!event) { return; } - if ( - this.turnState.status === "stopping" && - getOpenCodeEventSessionId(event) === this.sessionId - ) { - // Residue of the canceled run must not surface as a new turn. Its terminal - // is the authoritative end of the stop, so anything OpenCode publishes - // afterwards belongs to a new run by construction and takes the live path. - if (isOpenCodeTerminalEvent(event, this.sessionId)) { - this.finishStoppingTurn(this.turnState.stop); - } - this.traceOpenCode("provider.opencode.event.skip", { - n: eventCount, - reason: "turn_stopping", - type: event.type, - }); + this.observeRunnerStatusEvent(event); + if (this.discardEventWhileStopping(event, eventCount)) { return; } const translated = await this.translateEvent(event); @@ -3683,7 +3713,7 @@ class OpenCodeAgentSession implements AgentSession { foregroundEvents.push(translatedEvent); } } - if (!turnId && this.shouldStartAutonomousTurn(event, foregroundEvents)) { + if (!turnId && this.shouldStartAutonomousTurn(event)) { turnId = this.startAutonomousTurn(); } if (!turnId) { @@ -3724,6 +3754,27 @@ class OpenCodeAgentSession implements AgentSession { } } + private discardEventWhileStopping(event: OpenCodeEvent, eventCount: number): boolean { + if ( + this.turnState.status !== "stopping" || + getOpenCodeEventSessionId(event) !== this.sessionId + ) { + return false; + } + // Residue of the canceled run must not surface as a new turn. Its terminal + // is the authoritative end of the stop, so anything OpenCode publishes + // afterwards belongs to a new run by construction and takes the live path. + if (isOpenCodeTerminalEvent(event, this.sessionId)) { + this.finishStoppingTurn(this.turnState.stop); + } + this.traceOpenCode("provider.opencode.event.skip", { + n: eventCount, + reason: "turn_stopping", + type: event.type, + }); + return true; + } + private emitBackgroundPermissionRequests(events: readonly AgentStreamEvent[]): void { for (const event of events) { if (event.type === "permission_requested") { @@ -3732,37 +3783,20 @@ class OpenCodeAgentSession implements AgentSession { } } - private shouldStartAutonomousTurn( - event: OpenCodeEvent, - foregroundEvents: readonly AgentStreamEvent[], - ): boolean { + private observeRunnerStatusEvent(event: OpenCodeEvent): void { + if (getOpenCodeRunnerStatusFromEvent(event, this.sessionId) !== null) { + this.runnerStatusRevision += 1; + } + } + + private shouldStartAutonomousTurn(event: OpenCodeEvent): boolean { if (this.turnState.status !== "idle") { return false; } - if (getOpenCodeEventSessionId(event) !== this.sessionId) { - return false; - } - // OpenCode publishes the persisted user message before it marks the runner - // busy. That message is the earliest unambiguous boundary for a plugin- - // initiated parent turn; session metadata and assistant echoes are not. - if (event.type === "message.updated" && event.properties.info.role === "user") { - if (this.emittedUserMessageIds.has(event.properties.info.id)) { - return false; - } - return true; - } - if (!this.externallyDriven) { - return false; - } - if ( - foregroundEvents.some( - (foregroundEvent) => - foregroundEvent.type !== "thread_started" && !toTerminalTurnEvent(foregroundEvent), - ) - ) { - return true; - } - return event.type === "session.status" && event.properties.status.type === "busy"; + // Message records are mutable and can be patched after the runner stops. + // Only OpenCode's execution status is authoritative for autonomous activity. + const runnerStatus = getOpenCodeRunnerStatusFromEvent(event, this.sessionId); + return runnerStatus !== null && isOpenCodeRunnerActive(runnerStatus); } private startAutonomousTurn(): string { diff --git a/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts index db3fb4482..a3991cab0 100644 --- a/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts @@ -99,6 +99,7 @@ test("OpenCode remains idle while an OMO child works and completes the autonomou expect(childRunningIndex).toBeGreaterThan(rootRunningIndex); expect(rootIdleIndex).toBeGreaterThan(childRunningIndex); expect(childCompletedIndex).toBeGreaterThan(rootIdleIndex); + expect(wakeStartIndex).toBeGreaterThan(childCompletedIndex); expect(wakeStartIndex).toBeGreaterThan(-1); expect(autonomousTerminalIndex).toBeGreaterThan(wakeStartIndex); expect(cancellationAfterWake).toBe(false);