diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index b4c319e8e9a96..ecb05a5a02a15 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -3288,6 +3288,55 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("rejects unreadable rollback history before cutting the native session", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + out("message.list", { sessionID: SESSION, limit: "100", type: undefined }), + reply("message.list", { + data: [ + { id: "msg_kept", time: { created: 1 }, text: "first", type: "user" }, + { + id: "msg_compaction", + time: { created: 2 }, + type: "compaction", + status: "completed", + reason: "auto", + summary: "First turn summary", + }, + { id: "msg_dropped", time: { created: 3 }, text: "second", type: "user" }, + ], + cursor: {}, + }), + ]); + const now = yield* DateTime.now; + const turns = ["msg_kept", "msg_dropped"].map((nativeId, index) => ({ + id: ProviderTurnId.make(`provider-turn:rollback-history:${index}`), + providerThreadId: thread.id, + nodeId: NodeId.make(`node:rollback-history:${index}`), + runAttemptId: RunAttemptId.make(`attempt:rollback-history:${index}`), + nativeTurnRef: { driver: OPENCODE_PROVIDER, nativeId, strength: "weak" as const }, + ordinal: index + 1, + status: "completed" as const, + startedAt: now, + completedAt: now, + })); + const failure = yield* runtime + .rollbackThread({ + providerThread: thread, + target: { + type: "provider_turn", + checkpointId: CheckpointId.make("checkpoint:rollback-history"), + appRunOrdinal: 1, + providerTurn: turns[0]!, + }, + providerThreadTurns: turns, + }) + .pipe(Effect.flip); + assert.equal(failure._tag, "ProviderAdapterRollbackThreadError"); + assert.include(Cause.pretty(Cause.fail(failure)), '["recent"]'); + }).pipe(Effect.scoped), + ); + it.effect("clears a staged revert whose commit failed, before the next prompt too", () => Effect.gen(function* () { const kept = "msg_t3_turn_run-attempt:kept:1"; @@ -4075,8 +4124,22 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Effect.gen(function* () { const FORK = "ses_f1484db83ffeLGtrRCFimo1H0e"; const { runtime, thread, first, second } = yield* continued([ - out("message.list", ""), - reply("message.list", continuedHistory), + out("message.list", { sessionID: SESSION, limit: "100", type: "user" }), + reply("message.list", { data: [continuedHistory.data[1]], cursor: {} }), + out("message.list", { sessionID: SESSION, limit: "100", type: "synthetic" }), + reply("message.list", { + data: [ + { id: "msg_later_report", time: { created: 3 }, text: "later", type: "synthetic" }, + ], + cursor: { next: "synthetic-page-2" }, + }), + out("message.list", { + sessionID: SESSION, + limit: "100", + type: "synthetic", + cursor: "synthetic-page-2", + }), + reply("message.list", { data: [continuedHistory.data[0]], cursor: {} }), out("session.fork", { sessionID: SESSION, before: "msg_report" }), replyData("session.fork", sessionInfo({ id: FORK })), out("session.update", { sessionID: FORK, permissions: "" }), diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_fork/opencode_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_fork/opencode_transcript.ndjson index 57cbd14853460..a483f04813b31 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_fork/opencode_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/opencode2_fork/opencode_transcript.ndjson @@ -41,10 +41,12 @@ {"type": "emit_inbound", "label": "session.step.ended.2", "frame": {"type": "sdk.event", "event": {"id": "evt_0eb7b2461001qEeIO2dKdf1y1Y", "created": 1790657111137, "type": "session.step.ended", "location": {"directory": ""}, "data": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "assistantMessageID": "msg_0eb7b1fb30011jHuDOQ5YiY6eH", "finish": "stop", "rawFinish": "stop", "cost": 0, "tokens": {"input": 16, "output": 3, "reasoning": 0, "cache": {"read": 9327, "write": 0}}, "snapshot": "99482b54df6361365fa0ed7363ccf07edaa7fc93", "files": []}, "durable": {"aggregateID": "ses_f1484ea8affe5aTo8CcgbGIszj", "seq": 20, "version": 1}}}} {"type": "emit_inbound", "label": "session.usage.updated.2", "frame": {"type": "sdk.event", "event": {"id": "evt_0eb7b2465001TtO12Ofsv2KZTq", "created": 1790657111141, "type": "session.usage.updated", "data": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "cost": 0, "tokens": {"input": 8759, "output": 4, "reasoning": 99, "cache": {"read": 9813, "write": 0}}}}}} {"type": "emit_inbound", "label": "session.execution.succeeded.2", "frame": {"type": "sdk.event", "event": {"id": "evt_0eb7b24660011TVz6Q64I8aAQ7", "created": 1790657111142, "type": "session.execution.succeeded", "data": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj"}, "durable": {"aggregateID": "ses_f1484ea8affe5aTo8CcgbGIszj", "seq": 21, "version": 1}}}} -{"type": "expect_outbound", "label": "message.list", "frame": {"type": "message.list", "input": ""}} +{"type": "expect_outbound", "label": "message.list", "frame": {"type": "message.list", "input": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "limit": "100", "type": "user"}}} {"type": "emit_inbound", "label": "message.list.response", "frame": {"type": "sdk.response", "operation": "message.list", "data": {"data": [{"id": "msg_0eb7b1f9f001g95Bduh3bi8syS", "time": {"created": 1790657109932}, "text": "Reply exactly TWO.", "type": "user"}, {"id": "msg_0eb7b1597001e14VBV3nmQYdFS", "time": {"created": 1790657107378}, "text": "Reply exactly ONE.", "type": "user"}], "cursor": {"next": "cursor-after-first-page"}}}} -{"type": "expect_outbound", "label": "message.list.2", "frame": {"type": "message.list", "input": ""}} +{"type": "expect_outbound", "label": "message.list.2", "frame": {"type": "message.list", "input": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "limit": "100", "type": "user", "cursor": "cursor-after-first-page"}}} {"type": "emit_inbound", "label": "message.list.response.2", "frame": {"type": "sdk.response", "operation": "message.list", "data": {"data": [], "cursor": {}}}} +{"type": "expect_outbound", "label": "message.list.synthetic", "frame": {"type": "message.list", "input": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "limit": "100", "type": "synthetic"}}} +{"type": "emit_inbound", "label": "message.list.synthetic.response", "frame": {"type": "sdk.response", "operation": "message.list", "data": {"data": [], "cursor": {}}}} {"type": "expect_outbound", "label": "session.fork", "frame": {"type": "session.fork", "input": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "before": "msg_0eb7b1f9f001g95Bduh3bi8syS"}}} {"type": "emit_inbound", "label": "session.fork.response", "frame": {"type": "sdk.response", "operation": "session.fork", "data": {"data": {"id": "ses_f1484db83ffeLGtrRCFimo1H0e", "fork": {"sessionID": "ses_f1484ea8affe5aTo8CcgbGIszj", "boundary": {"type": "before", "messageID": "msg_0eb7b1f9f001g95Bduh3bi8syS"}}, "projectID": "global", "model": {"id": "big-pickle", "providerID": "opencode", "variant": "default"}, "cost": 0, "tokens": {"input": 0, "output": 0, "reasoning": 0, "cache": {"read": 0, "write": 0}}, "time": {"created": 1790657111165, "updated": 1790657111165}, "title": "spike fork (fork #1)", "location": {"directory": ""}, "permissions": [{"action": "*", "resource": "*", "effect": "allow"}, {"action": "subagent", "resource": "*", "effect": "deny"}]}}}} {"type": "expect_outbound", "label": "session.update", "frame": {"type": "session.update", "input": {"sessionID": "ses_f1484db83ffeLGtrRCFimo1H0e", "permissions": [{"action": "*", "resource": "*", "effect": "allow"}, {"action": "t3-code-*", "resource": "*", "effect": "deny"}, {"action": "t3-code-thread_opencode2-fork_target_*", "resource": "*", "effect": "allow"}]}}} diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index f49de61b24690..40fdb20ff4d8d 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -2925,12 +2925,19 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); }); - // The API accepts one message type; boundaries require both user and synthetic messages. - const userMessages = (sessionId: string) => - OpenCode2Client.paginate( - { sessionID: Session.ID.make(sessionId), limit: 100 }, - client.message.list, - ).pipe( + // Rollback must validate full history before native mutation; forks only need prompts. + const userMessages = (sessionId: string, types?: ReadonlyArray<"user" | "synthetic">) => + Stream.fromIterable(types ?? [undefined]).pipe( + Stream.flatMap((type) => + OpenCode2Client.paginate( + { + sessionID: Session.ID.make(sessionId), + limit: 100, + ...(type === undefined ? {} : { type }), + }, + client.message.list, + ), + ), Stream.runCollect, Effect.map( (messages) => @@ -4008,7 +4015,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr forkInput.sourceProviderTurns ?? [], source.id, selected, - yield* userMessages(sourceId), + yield* userMessages(sourceId, ["user", "synthetic"]), ); const forked = yield* client.session.fork({ sessionID: Session.ID.make(sourceId),