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

Use OpenCode busy status for background activity (#2696)

Browse Source 
* fix(server): ignore OpenCode snapshot diffs when starting autonomous turns

OpenCode stamps its filesystem-snapshot diff onto the user message after
the turn goes idle. Since #2258 Paseo treats an unknown idle user
message.updated as the start of an autonomous turn, so that late metadata
write opened a turn no provider-idle boundary would ever close. The agent
stayed falsely running until the next submission interrupted it.

Reject user message.updated events carrying summary.diffs. Real prompts
never carry snapshot diffs, so genuine autonomous turns still start, and
filesystem snapshots stay enabled for undo and revert.

* fix(server): trust OpenCode busy status for autonomous activity

Mutable message records can receive metadata after execution ends. Treat the provider's busy status as the activity boundary and buffer preceding content until that boundary arrives.

* test(server): cover autonomous provider activity

* fix(server): reconcile OpenCode wakes from busy status

# Conflicts:
#	packages/server/src/server/agent/providers/opencode-agent.ts

* fix(server): preserve background activity boundary after stops

* fix(server): keep idle events outside background activity

Only provider execution status proves a new activity boundary. Reconcile adopted sessions once at subscription time instead of retaining unowned idle events for a future run.

* fix(server): order adopted session reconciliation

Live runner-status events supersede an in-flight status snapshot. Treat busy and retry as one active provider state.

* test(server): remove misleading snapshot gate

---------

Co-authored-by: Des Flynn <des@desflynn.com>
This commit is contained in:
Mohamed Boudra
2026-07-31 13:58:08 +02:00
committed by GitHub
co-authored by Des Flynn
parent 8e983c5d27
commit d9b72e1c76
3 changed files with 327 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/opencode-agent.test.ts
+240 -8
View File
@@ -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<void>();
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<ReturnType<OpenCodeAgentClient["createSession"]>>;
@@ -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<void>();
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<void>();
const releaseStatus = createTestDeferred<void>();
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<void>();
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<void>((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: "<system-reminder>All background tasks are complete.</system-reminder>",
}),
}),
);
} 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<void> {
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();
});
});
packages/server/src/server/agent/providers/opencode-agent.ts
+86 -52
View File
@@ -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<void> = Promise.resolve();
private externalStatusReconciliationStarted = false;
private runnerStatusRevision = 0;
private readonly runningToolCalls = new Map<string, ToolCallTimelineItem>();
private subAgentsByCallId = new Map<string, OpenCodeSubAgentActivityState>();
private subAgentCallIdByChildSessionId = new Map<string, string>();
@@ -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<void> {
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 {
packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts
+1
View File
@@ -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);