From a72992e00f528b213ad67ad56db4a77eea7bedf1 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 9 Jul 2026 17:25:40 -0400 Subject: [PATCH] feat(core): session.pending.list API with pending-only session_pending storage (#36126) --- AGENTS.md | 4 +- CONTEXT.md | 2 +- bun.lock | 2 + packages/client/src/effect/api/api.ts | 78 ++--- .../client/src/effect/generated/client.ts | 77 +++-- packages/client/src/effect/index.ts | 2 +- .../client/src/promise/generated/client.ts | 15 + .../client/src/promise/generated/types.ts | 50 ++- .../client/test/contract-identity.test.ts | 44 --- packages/client/test/promise.test.ts | 28 ++ packages/core/schema.json | 306 ++++++++---------- packages/core/src/database/database.ts | 8 +- packages/core/src/database/migration.gen.ts | 1 + .../20260709190621_session_pending_table.ts | 35 ++ packages/core/src/database/schema.gen.ts | 46 ++- packages/core/src/session.ts | 46 +-- .../core/src/session/{input.ts => pending.ts} | 254 ++++++++------- packages/core/src/session/projector.ts | 44 ++- packages/core/src/session/runner/llm.ts | 34 +- packages/core/src/session/sql.ts | 27 +- packages/core/test/database-migration.test.ts | 82 ++++- packages/core/test/session-compact.test.ts | 4 +- packages/core/test/session-create.test.ts | 52 ++- packages/core/test/session-projector.test.ts | 19 +- packages/core/test/session-prompt.test.ts | 99 ++++-- packages/core/test/session-runner.test.ts | 44 ++- packages/core/test/shared-schema.test.ts | 14 +- packages/core/test/tool-subagent.test.ts | 4 +- .../test/server/httpapi-exercise/index.ts | 8 + .../test/server/httpapi-session.test.ts | 17 +- packages/protocol/src/groups/session.ts | 34 +- packages/schema/src/index.ts | 2 +- packages/schema/src/session-event.ts | 4 +- .../{session-input.ts => session-pending.ts} | 29 +- packages/schema/test/contract-hygiene.test.ts | 14 +- packages/sdk-next/package.json | 2 + packages/sdk-next/src/index.ts | 2 +- .../sdk-next/test/contract-identity.test.ts | 47 ++- packages/sdk-next/test/embedded.test.ts | 8 + packages/sdk/js/src/v2/gen/sdk.gen.ts | 32 ++ packages/sdk/js/src/v2/gen/types.gen.ts | 109 ++++--- packages/server/src/handlers/session.ts | 17 + specs/v2/schema-changelog.md | 13 + 43 files changed, 1048 insertions(+), 711 deletions(-) create mode 100644 packages/core/src/database/migration/20260709190621_session_pending_table.ts rename packages/core/src/session/{input.ts => pending.ts} (58%) rename packages/schema/src/{session-input.ts => session-pending.ts} (75%) diff --git a/AGENTS.md b/AGENTS.md index d6bd72d4b0..d2ad3f4a78 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -154,8 +154,8 @@ const table = sqliteTable("session", { ## V2 Session Core - Keep durable events minimal: record irreducible new facts and do not repeat state derivable by folding the ordered aggregate history. Enrich projections and read models with previous or derived state when consumers need self-contained views. -- Keep durable prompt admission separate from model execution. `SessionV2.prompt(...)` admits one durable `session_input` row before scheduling advisory `SessionExecution.wake(sessionID)` unless `resume: false` requests admit-only behavior. The serialized runner promotes admitted inputs into visible user messages at safe boundaries. -- Reusing a Session ID adopts the existing Session. Reusing a prompt message ID reconciles an exact retry only when Session, prompt, and delivery mode match; conflicting reuse fails. Historical projected prompts lazily synthesize promoted inbox records during exact retry. +- Keep durable prompt admission separate from model execution. `SessionV2.prompt(...)` admits one durable `session_pending` row before scheduling advisory `SessionExecution.wake(sessionID)` unless `resume: false` requests admit-only behavior. The serialized runner promotes admitted inputs into visible user messages at safe boundaries, consuming the pending row in the same event transaction; `session_pending` stores only unconsumed work. +- Reusing a Session ID adopts the existing Session. Reusing a prompt message ID reconciles an exact retry only when Session, prompt, and delivery mode match; conflicting reuse fails. Retry of an already-promoted input reconciles against the projected message and the durable admitted event rather than a retained row. - Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and discovers placement through `SessionStore` plus `LocationServiceMap.get(session.location)` only when a drain starts; no layer should take a Session ID. V2 interruption targets the active process-local ownership chain for that Session; idle or missing interruption is a no-op. - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics. - Preserve one explicit `llm.stream(request)` call per step and reload projected history before durable continuation. Do not bridge through legacy `SessionPrompt.loop(...)` or delegate orchestration to an in-memory tool loop. diff --git a/CONTEXT.md b/CONTEXT.md index ce880380d1..e1f2b2c36a 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -198,7 +198,7 @@ _Avoid_: Response envelope - `sessions.context({ sessionID })` preserves the existing message-only operation. It returns projected **Session History**; it does not include or represent the complete **Model Context**, whose system text, **Instruction Baseline**, tools, and step-local additions remain separate. - **Open question**: Should a future, separately named operation expose complete **Model Context**, including the instruction baseline, applied instruction metadata, tools, and step-local additions? - `sessions.prompt(...)` exposes `resume?: boolean`. Omitting it preserves durable admission followed by an advisory execution wake; `resume: false` requests durable admit-only behavior. -- The public operation remains `sessions.prompt(...)`; `SessionInput.admit` is the internal primitive, while the public `Admission` result and `resume` option express its durable admission semantics. +- The public operation remains `sessions.prompt(...)`; `SessionPending.admit` is the internal primitive, while the public `Admission` result and `resume` option express its durable admission semantics. - `sessions.create(...)` accepts an optional `location`. Omission resolves through the connected OpenCode instance's default or current location; an explicit value selects a known location. Networked and embedded transports use the same handler semantics. - `sessions.switchAgent({ sessionID, agent })` is part of the common client alongside `sessions.switchModel(...)`. It affects subsequent Session activity and fails with `SessionNotFoundError` for an unknown Session. - The **Embedded OpenCode** Layer delegates to the same scoped creation path; it does not define a second implementation. diff --git a/bun.lock b/bun.lock index 707302aa91..94ebdae5cd 100644 --- a/bun.lock +++ b/bun.lock @@ -790,6 +790,8 @@ "effect": "catalog:", }, "devDependencies": { + "@opencode-ai/httpapi-codegen": "workspace:*", + "@opencode-ai/protocol": "workspace:*", "@tsconfig/bun": "catalog:", "@types/bun": "catalog:", "@typescript/native-preview": "catalog:", diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index d1f726d077..4eedd8b15b 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -226,64 +226,69 @@ export type Endpoint5_20Input = { readonly sessionID: Endpoint5_20Request["param export type Endpoint5_20Output = EffectValue>["data"] export type SessionContextOperation = (input: Endpoint5_20Input) => Effect.Effect -type Endpoint5_21Request = Parameters[0] +type Endpoint5_21Request = Parameters[0] export type Endpoint5_21Input = { readonly sessionID: Endpoint5_21Request["params"]["sessionID"] } -export type Endpoint5_21Output = EffectValue< +export type Endpoint5_21Output = EffectValue>["data"] +export type SessionPendingListOperation = (input: Endpoint5_21Input) => Effect.Effect + +type Endpoint5_22Request = Parameters[0] +export type Endpoint5_22Input = { readonly sessionID: Endpoint5_22Request["params"]["sessionID"] } +export type Endpoint5_22Output = EffectValue< ReturnType >["data"] export type SessionInstructionsEntryListOperation = ( - input: Endpoint5_21Input, -) => Effect.Effect - -type Endpoint5_22Request = Parameters[0] -export type Endpoint5_22Input = { - readonly sessionID: Endpoint5_22Request["params"]["sessionID"] - readonly key: Endpoint5_22Request["params"]["key"] - readonly value: Endpoint5_22Request["payload"]["value"] -} -export type Endpoint5_22Output = EffectValue> -export type SessionInstructionsEntryPutOperation = ( input: Endpoint5_22Input, ) => Effect.Effect -type Endpoint5_23Request = Parameters[0] +type Endpoint5_23Request = Parameters[0] export type Endpoint5_23Input = { readonly sessionID: Endpoint5_23Request["params"]["sessionID"] readonly key: Endpoint5_23Request["params"]["key"] + readonly value: Endpoint5_23Request["payload"]["value"] } -export type Endpoint5_23Output = EffectValue< - ReturnType -> -export type SessionInstructionsEntryRemoveOperation = ( +export type Endpoint5_23Output = EffectValue> +export type SessionInstructionsEntryPutOperation = ( input: Endpoint5_23Input, ) => Effect.Effect -type Endpoint5_24Request = Parameters[0] +type Endpoint5_24Request = Parameters[0] export type Endpoint5_24Input = { readonly sessionID: Endpoint5_24Request["params"]["sessionID"] - readonly after?: Endpoint5_24Request["query"]["after"] - readonly follow?: Endpoint5_24Request["query"]["follow"] + readonly key: Endpoint5_24Request["params"]["key"] } -export type Endpoint5_24Output = StreamValue>> -export type SessionLogOperation = (input: Endpoint5_24Input) => Stream.Stream +export type Endpoint5_24Output = EffectValue< + ReturnType +> +export type SessionInstructionsEntryRemoveOperation = ( + input: Endpoint5_24Input, +) => Effect.Effect -type Endpoint5_25Request = Parameters[0] -export type Endpoint5_25Input = { readonly sessionID: Endpoint5_25Request["params"]["sessionID"] } -export type Endpoint5_25Output = EffectValue> -export type SessionInterruptOperation = (input: Endpoint5_25Input) => Effect.Effect +type Endpoint5_25Request = Parameters[0] +export type Endpoint5_25Input = { + readonly sessionID: Endpoint5_25Request["params"]["sessionID"] + readonly after?: Endpoint5_25Request["query"]["after"] + readonly follow?: Endpoint5_25Request["query"]["follow"] +} +export type Endpoint5_25Output = StreamValue>> +export type SessionLogOperation = (input: Endpoint5_25Input) => Stream.Stream -type Endpoint5_26Request = Parameters[0] +type Endpoint5_26Request = Parameters[0] export type Endpoint5_26Input = { readonly sessionID: Endpoint5_26Request["params"]["sessionID"] } -export type Endpoint5_26Output = EffectValue> -export type SessionBackgroundOperation = (input: Endpoint5_26Input) => Effect.Effect +export type Endpoint5_26Output = EffectValue> +export type SessionInterruptOperation = (input: Endpoint5_26Input) => Effect.Effect -type Endpoint5_27Request = Parameters[0] -export type Endpoint5_27Input = { - readonly sessionID: Endpoint5_27Request["params"]["sessionID"] - readonly messageID: Endpoint5_27Request["params"]["messageID"] +type Endpoint5_27Request = Parameters[0] +export type Endpoint5_27Input = { readonly sessionID: Endpoint5_27Request["params"]["sessionID"] } +export type Endpoint5_27Output = EffectValue> +export type SessionBackgroundOperation = (input: Endpoint5_27Input) => Effect.Effect + +type Endpoint5_28Request = Parameters[0] +export type Endpoint5_28Input = { + readonly sessionID: Endpoint5_28Request["params"]["sessionID"] + readonly messageID: Endpoint5_28Request["params"]["messageID"] } -export type Endpoint5_27Output = EffectValue>["data"] -export type SessionMessageOperation = (input: Endpoint5_27Input) => Effect.Effect +export type Endpoint5_28Output = EffectValue>["data"] +export type SessionMessageOperation = (input: Endpoint5_28Input) => Effect.Effect export interface SessionApi { readonly list: SessionListOperation @@ -309,6 +314,7 @@ export interface SessionApi { readonly commit: SessionRevertCommitOperation } readonly context: SessionContextOperation + readonly pending: { readonly list: SessionPendingListOperation } readonly instructions: { readonly entry: { readonly list: SessionInstructionsEntryListOperation diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 3b39da28a0..f11982abc3 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -318,43 +318,51 @@ const Endpoint5_20 = (raw: RawClient["server.session"]) => (input: Endpoint5_20I Effect.map((value) => value.data), ) -type Endpoint5_21Request = Parameters[0] +type Endpoint5_21Request = Parameters[0] type Endpoint5_21Input = { readonly sessionID: Endpoint5_21Request["params"]["sessionID"] } const Endpoint5_21 = (raw: RawClient["server.session"]) => (input: Endpoint5_21Input) => + raw["session.pending.list"]({ params: { sessionID: input["sessionID"] } }).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ) + +type Endpoint5_22Request = Parameters[0] +type Endpoint5_22Input = { readonly sessionID: Endpoint5_22Request["params"]["sessionID"] } +const Endpoint5_22 = (raw: RawClient["server.session"]) => (input: Endpoint5_22Input) => raw["session.instructions.entry.list"]({ params: { sessionID: input["sessionID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), ) -type Endpoint5_22Request = Parameters[0] -type Endpoint5_22Input = { - readonly sessionID: Endpoint5_22Request["params"]["sessionID"] - readonly key: Endpoint5_22Request["params"]["key"] - readonly value: Endpoint5_22Request["payload"]["value"] +type Endpoint5_23Request = Parameters[0] +type Endpoint5_23Input = { + readonly sessionID: Endpoint5_23Request["params"]["sessionID"] + readonly key: Endpoint5_23Request["params"]["key"] + readonly value: Endpoint5_23Request["payload"]["value"] } -const Endpoint5_22 = (raw: RawClient["server.session"]) => (input: Endpoint5_22Input) => +const Endpoint5_23 = (raw: RawClient["server.session"]) => (input: Endpoint5_23Input) => raw["session.instructions.entry.put"]({ params: { sessionID: input["sessionID"], key: input["key"] }, payload: { value: input["value"] }, }).pipe(Effect.mapError(mapClientError)) -type Endpoint5_23Request = Parameters[0] -type Endpoint5_23Input = { - readonly sessionID: Endpoint5_23Request["params"]["sessionID"] - readonly key: Endpoint5_23Request["params"]["key"] +type Endpoint5_24Request = Parameters[0] +type Endpoint5_24Input = { + readonly sessionID: Endpoint5_24Request["params"]["sessionID"] + readonly key: Endpoint5_24Request["params"]["key"] } -const Endpoint5_23 = (raw: RawClient["server.session"]) => (input: Endpoint5_23Input) => +const Endpoint5_24 = (raw: RawClient["server.session"]) => (input: Endpoint5_24Input) => raw["session.instructions.entry.remove"]({ params: { sessionID: input["sessionID"], key: input["key"] } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint5_24Request = Parameters[0] -type Endpoint5_24Input = { - readonly sessionID: Endpoint5_24Request["params"]["sessionID"] - readonly after?: Endpoint5_24Request["query"]["after"] - readonly follow?: Endpoint5_24Request["query"]["follow"] +type Endpoint5_25Request = Parameters[0] +type Endpoint5_25Input = { + readonly sessionID: Endpoint5_25Request["params"]["sessionID"] + readonly after?: Endpoint5_25Request["query"]["after"] + readonly follow?: Endpoint5_25Request["query"]["follow"] } -const Endpoint5_24 = (raw: RawClient["server.session"]) => (input: Endpoint5_24Input) => +const Endpoint5_25 = (raw: RawClient["server.session"]) => (input: Endpoint5_25Input) => Stream.unwrap( raw["session.log"]({ params: { sessionID: input["sessionID"] }, @@ -365,22 +373,22 @@ const Endpoint5_24 = (raw: RawClient["server.session"]) => (input: Endpoint5_24I ), ) -type Endpoint5_25Request = Parameters[0] -type Endpoint5_25Input = { readonly sessionID: Endpoint5_25Request["params"]["sessionID"] } -const Endpoint5_25 = (raw: RawClient["server.session"]) => (input: Endpoint5_25Input) => - raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) - -type Endpoint5_26Request = Parameters[0] +type Endpoint5_26Request = Parameters[0] type Endpoint5_26Input = { readonly sessionID: Endpoint5_26Request["params"]["sessionID"] } const Endpoint5_26 = (raw: RawClient["server.session"]) => (input: Endpoint5_26Input) => + raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) + +type Endpoint5_27Request = Parameters[0] +type Endpoint5_27Input = { readonly sessionID: Endpoint5_27Request["params"]["sessionID"] } +const Endpoint5_27 = (raw: RawClient["server.session"]) => (input: Endpoint5_27Input) => raw["session.background"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint5_27Request = Parameters[0] -type Endpoint5_27Input = { - readonly sessionID: Endpoint5_27Request["params"]["sessionID"] - readonly messageID: Endpoint5_27Request["params"]["messageID"] +type Endpoint5_28Request = Parameters[0] +type Endpoint5_28Input = { + readonly sessionID: Endpoint5_28Request["params"]["sessionID"] + readonly messageID: Endpoint5_28Request["params"]["messageID"] } -const Endpoint5_27 = (raw: RawClient["server.session"]) => (input: Endpoint5_27Input) => +const Endpoint5_28 = (raw: RawClient["server.session"]) => (input: Endpoint5_28Input) => raw["session.message"]({ params: { sessionID: input["sessionID"], messageID: input["messageID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), @@ -406,11 +414,12 @@ const adaptGroup5 = (raw: RawClient["server.session"]) => ({ wait: Endpoint5_16(raw), revert: { stage: Endpoint5_17(raw), clear: Endpoint5_18(raw), commit: Endpoint5_19(raw) }, context: Endpoint5_20(raw), - instructions: { entry: { list: Endpoint5_21(raw), put: Endpoint5_22(raw), remove: Endpoint5_23(raw) } }, - log: Endpoint5_24(raw), - interrupt: Endpoint5_25(raw), - background: Endpoint5_26(raw), - message: Endpoint5_27(raw), + pending: { list: Endpoint5_21(raw) }, + instructions: { entry: { list: Endpoint5_22(raw), put: Endpoint5_23(raw), remove: Endpoint5_24(raw) } }, + log: Endpoint5_25(raw), + interrupt: Endpoint5_26(raw), + background: Endpoint5_27(raw), + message: Endpoint5_28(raw), }) type Endpoint6_0Request = Parameters[0] diff --git a/packages/client/src/effect/index.ts b/packages/client/src/effect/index.ts index 3e47320953..67c2c42142 100644 --- a/packages/client/src/effect/index.ts +++ b/packages/client/src/effect/index.ts @@ -38,7 +38,7 @@ export { Question } from "@opencode-ai/schema/question" export { Reference } from "@opencode-ai/schema/reference" export { AbsolutePath, RelativePath } from "@opencode-ai/schema/schema" export { Session } from "@opencode-ai/schema/session" -export { SessionInput } from "@opencode-ai/schema/session-input" +export { SessionPending } from "@opencode-ai/schema/session-pending" export { SessionMessage } from "@opencode-ai/schema/session-message" export { Skill } from "@opencode-ai/schema/skill" export { Prompt } from "@opencode-ai/schema/prompt" diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 606673944c..3982a99566 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -48,6 +48,8 @@ import type { SessionRevertCommitOutput, SessionContextInput, SessionContextOutput, + SessionPendingListInput, + SessionPendingListOutput, SessionInstructionsEntryListInput, SessionInstructionsEntryListOutput, SessionInstructionsEntryPutInput, @@ -666,6 +668,19 @@ export function make(options: ClientOptions) { }, requestOptions, ).then((value) => value.data), + pending: { + list: (input: SessionPendingListInput, requestOptions?: RequestOptions) => + request<{ readonly data: SessionPendingListOutput }>( + { + method: "GET", + path: `/api/session/${encodeURIComponent(input.sessionID)}/pending`, + successStatus: 200, + declaredStatuses: [404, 400, 401], + empty: false, + }, + requestOptions, + ).then((value) => value.data), + }, instructions: { entry: { list: (input: SessionInstructionsEntryListInput, requestOptions?: RequestOptions) => diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 0ae6093ba6..3277dcdccf 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -624,7 +624,6 @@ export type SessionPromptOutput = { id: string sessionID: string timeCreated: number - promotedSeq?: number type: "user" data: { text: string @@ -824,7 +823,6 @@ export type SessionCommandOutput = { id: string sessionID: string timeCreated: number - promotedSeq?: number type: "user" data: { text: string @@ -922,7 +920,6 @@ export type SessionSyntheticOutput = { id: string sessionID: string timeCreated: number - promotedSeq?: number type: "synthetic" data: { text: string; description?: string; metadata?: { [x: string]: JsonValue } } delivery: "steer" | "queue" @@ -943,14 +940,7 @@ export type SessionCompactInput = { } export type SessionCompactOutput = { - data: { - admittedSeq: number - id: string - sessionID: string - timeCreated: number - type: "compaction" - handledSeq?: number - } + data: { admittedSeq: number; id: string; sessionID: string; timeCreated: number; type: "compaction" } }["data"] export type SessionWaitInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } @@ -1145,6 +1135,44 @@ export type SessionContextOutput = { > }["data"] +export type SessionPendingListInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } + +export type SessionPendingListOutput = { + data: Array< + | { + admittedSeq: number + id: string + sessionID: string + timeCreated: number + type: "user" + data: { + text: string + files?: Array<{ + data: string + mime: string + source: { type: "inline" } | { type: "uri"; uri: string } + name?: string + description?: string + mention?: { start: number; end: number; text: string } + }> + agents?: Array<{ name: string; mention?: { start: number; end: number; text: string } }> + metadata?: { [x: string]: JsonValue } + } + delivery: "steer" | "queue" + } + | { + admittedSeq: number + id: string + sessionID: string + timeCreated: number + type: "synthetic" + data: { text: string; description?: string; metadata?: { [x: string]: JsonValue } } + delivery: "steer" | "queue" + } + | { admittedSeq: number; id: string; sessionID: string; timeCreated: number; type: "compaction" } + > +}["data"] + export type SessionInstructionsEntryListInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } export type SessionInstructionsEntryListOutput = { data: Array<{ key: string; value: JsonValue }> }["data"] diff --git a/packages/client/test/contract-identity.test.ts b/packages/client/test/contract-identity.test.ts index 3b6c1b82b7..8de2115fa7 100644 --- a/packages/client/test/contract-identity.test.ts +++ b/packages/client/test/contract-identity.test.ts @@ -1,25 +1,10 @@ import { expect, test } from "bun:test" import { Schema } from "effect" -import { AgentV2 } from "@opencode-ai/core/agent" -import { Location as CoreLocation } from "@opencode-ai/core/location" -import { ModelV2 } from "@opencode-ai/core/model" -import { ProjectV2 } from "@opencode-ai/core/project" -import { SessionV2 } from "@opencode-ai/core/session" -import { SessionInput as CoreSessionInput } from "@opencode-ai/core/session/input" -import { SessionMessage as CoreSessionMessage } from "@opencode-ai/core/session/message" import { Agent } from "@opencode-ai/schema/agent" -import { Location } from "@opencode-ai/schema/location" import { Model } from "@opencode-ai/schema/model" -import { Project } from "@opencode-ai/schema/project" -import { Provider } from "@opencode-ai/schema/provider" import { Prompt } from "@opencode-ai/schema/prompt" import { Session } from "@opencode-ai/schema/session" -import { SessionInput } from "@opencode-ai/schema/session-input" import { SessionMessage } from "@opencode-ai/schema/session-message" -import { Workspace } from "@opencode-ai/schema/workspace" -import { Api } from "@opencode-ai/server/api" -import { compile, emitPromise } from "@opencode-ai/httpapi-codegen" -import { ClientApi, groupNames, promiseOmitEndpoints } from "../src/contract" const Client = await import("../src/effect") @@ -29,34 +14,6 @@ test("effect entrypoint exposes canonical Schema contracts", () => { expect(Client.Session).toBe(Session) }) -test("Core and Server reuse the authoritative Schema and Protocol values", () => { - expect(AgentV2.ID).toBe(Agent.ID) - expect(CoreLocation.Ref).toBe(Location.Ref) - expect(ModelV2.Ref).toBe(Model.Ref) - expect(SessionV2.Info).toBe(Session.Info) - expect(ProjectV2.Current).toBe(Project.Current) - expect(ProjectV2.Directory).toBe(Project.Directory) - expect(ProjectV2.Directories).toBe(Project.Directories) - expect(CoreSessionInput.Message).toBe(SessionInput.Message) - expect(CoreSessionInput.User).toBe(SessionInput.User) - expect(CoreSessionInput.Synthetic).toBe(SessionInput.Synthetic) - expect(CoreSessionMessage.Info).toBe(SessionMessage.Info) - expect(Api.groups["server.session"].identifier).toBe("server.session") - expect(Api.groups["server.project"].identifier).toBe("server.project") - expect(Object.keys(ClientApi.groups)).toEqual(Object.keys(Api.groups)) - expect(Session.ID.create()).toStartWith("ses_") - expect(Project.ID.global).toBe("global") - expect(Provider.ID.anthropic).toBe("anthropic") - expect(Workspace.ID.create()).toStartWith("wrk_") -}) - -test("client and Server contracts generate identically", () => { - const server = compile(Api, { groupNames, omitEndpoints: promiseOmitEndpoints }) - const client = compile(ClientApi, { groupNames, omitEndpoints: promiseOmitEndpoints }) - - expect(emitPromise(client)).toEqual(emitPromise(server)) -}) - test("shared DTO schemas construct and decode plain objects", () => { const made = Prompt.make({ text: "hello" }) const decoded = Schema.decodeUnknownSync(Prompt)({ text: "hello" }) @@ -67,5 +24,4 @@ test("shared DTO schemas construct and decode plain objects", () => { expect(Object.getPrototypeOf(content)).toBe(Object.prototype) expect(Prompt.ast.annotations?.identifier).toBe("Prompt") expect(SessionMessage.AssistantText.ast.annotations?.identifier).toBe("Session.Message.Assistant.Text") - expect(CoreSessionMessage.AssistantText).toBe(SessionMessage.AssistantText) }) diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 015e9ebf5e..e29522c226 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -227,6 +227,34 @@ test("session instructions methods use the public HTTP contract", async () => { ]) }) +test("session.pending.list uses the public HTTP contract", async () => { + const requests: Array<{ method: string; url: string }> = [] + const pending = [ + { + admittedSeq: 3, + id: "msg_pending", + sessionID: "ses_test", + timeCreated: 1_717_171_717_000, + type: "user", + data: { text: "Fix the failing tests" }, + delivery: "steer", + }, + ] + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init) + requests.push({ method: request.method, url: request.url }) + return Response.json({ data: pending }) + }, + }) + + const result = await client.session.pending.list({ sessionID: "ses_test" }) + + expect(result).toEqual(pending) + expect(requests).toEqual([{ method: "GET", url: "http://localhost:3000/api/session/ses_test/pending" }]) +}) + test("event.subscribe exposes the Promise event stream wire projection", async () => { const client = OpenCode.make({ baseUrl: "http://localhost:3000", diff --git a/packages/core/schema.json b/packages/core/schema.json index 06cec4eaed..5d216eefc9 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "01451b27-1e51-4657-b2d0-4b457dffa3ec", - "prevIds": ["8c1748cc-f978-4df4-879d-fe7fda5c4c34"], + "id": "2c759c08-79c8-4179-9e15-d2fea45c9dec", + "prevIds": ["01451b27-1e51-4657-b2d0-4b457dffa3ec"], "ddl": [ { "name": "workspace", @@ -65,11 +65,11 @@ "entityType": "tables" }, { - "name": "session_input", + "name": "session_message", "entityType": "tables" }, { - "name": "session_message", + "name": "session_pending", "entityType": "tables" }, { @@ -980,86 +980,6 @@ "entityType": "columns", "table": "part" }, - { - "type": "text", - "notNull": false, - "autoincrement": false, - "default": null, - "generated": null, - "name": "id", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "text", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "session_id", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "text", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "type", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "text", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "data", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "text", - "notNull": false, - "autoincrement": false, - "default": null, - "generated": null, - "name": "delivery", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "integer", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "admitted_seq", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "integer", - "notNull": false, - "autoincrement": false, - "default": null, - "generated": null, - "name": "promoted_seq", - "entityType": "columns", - "table": "session_input" - }, - { - "type": "integer", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "time_created", - "entityType": "columns", - "table": "session_input" - }, { "type": "text", "notNull": false, @@ -1130,6 +1050,76 @@ "entityType": "columns", "table": "session_message" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "type", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "delivery", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "admitted_seq", + "entityType": "columns", + "table": "session_pending" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "session_pending" + }, { "type": "text", "notNull": false, @@ -1616,9 +1606,9 @@ "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, - "name": "fk_session_input_session_id_session_id_fk", + "name": "fk_session_message_session_id_session_id_fk", "entityType": "fks", - "table": "session_input" + "table": "session_message" }, { "columns": ["session_id"], @@ -1627,9 +1617,9 @@ "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, - "name": "fk_session_message_session_id_session_id_fk", + "name": "fk_session_input_session_id_session_id_fk", "entityType": "fks", - "table": "session_message" + "table": "session_pending" }, { "columns": ["project_id"], @@ -1761,15 +1751,15 @@ { "columns": ["id"], "nameExplicit": false, - "name": "session_input_pk", - "table": "session_input", + "name": "session_message_pk", + "table": "session_message", "entityType": "pks" }, { "columns": ["id"], "nameExplicit": false, - "name": "session_message_pk", - "table": "session_message", + "name": "session_input_pk", + "table": "session_pending", "entityType": "pks" }, { @@ -1902,82 +1892,6 @@ "entityType": "indexes", "table": "part" }, - { - "columns": [ - { - "value": "session_id", - "isExpression": false - }, - { - "value": "promoted_seq", - "isExpression": false - }, - { - "value": "delivery", - "isExpression": false - }, - { - "value": "admitted_seq", - "isExpression": false - } - ], - "isUnique": false, - "where": null, - "origin": "manual", - "name": "session_input_session_pending_delivery_seq_idx", - "entityType": "indexes", - "table": "session_input" - }, - { - "columns": [ - { - "value": "session_id", - "isExpression": false - } - ], - "isUnique": true, - "where": "\"session_input\".\"type\" = 'compaction' and \"session_input\".\"promoted_seq\" is null", - "origin": "manual", - "name": "session_input_session_pending_compaction_idx", - "entityType": "indexes", - "table": "session_input" - }, - { - "columns": [ - { - "value": "session_id", - "isExpression": false - }, - { - "value": "admitted_seq", - "isExpression": false - } - ], - "isUnique": true, - "where": null, - "origin": "manual", - "name": "session_input_session_admitted_seq_idx", - "entityType": "indexes", - "table": "session_input" - }, - { - "columns": [ - { - "value": "session_id", - "isExpression": false - }, - { - "value": "promoted_seq", - "isExpression": false - } - ], - "isUnique": true, - "where": null, - "origin": "manual", - "name": "session_input_session_promoted_seq_idx", - "entityType": "indexes", - "table": "session_input" - }, { "columns": [ { @@ -2054,6 +1968,60 @@ "entityType": "indexes", "table": "session_message" }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + }, + { + "value": "delivery", + "isExpression": false + }, + { + "value": "admitted_seq", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "session_pending_session_delivery_seq_idx", + "entityType": "indexes", + "table": "session_pending" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": true, + "where": "\"session_pending\".\"type\" = 'compaction'", + "origin": "manual", + "name": "session_pending_session_compaction_idx", + "entityType": "indexes", + "table": "session_pending" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + }, + { + "value": "admitted_seq", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "session_pending_session_admitted_seq_idx", + "entityType": "indexes", + "table": "session_pending" + }, { "columns": [ { @@ -2111,5 +2079,5 @@ "table": "session" } ], - "renames": [] + "renames": ["session_input->session_pending"] } diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index f3cc8b3f9f..e6c36d75b8 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -54,4 +54,10 @@ export function path() { return join(Global.Path.data, `opencode-${InstallationChannel.replace(/[^a-zA-Z0-9._-]/g, "-")}.db`) } -export const node = makeGlobalNode({ service: Service, layer: layerFromPath(path()), deps: [] }) +// Resolve the database path lazily so tests and embedders that set +// Flag.OPENCODE_DB after module evaluation still control the storage target. +export const node = makeGlobalNode({ + service: Service, + layer: Layer.suspend(() => layerFromPath(path())), + deps: [], +}) diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 9f5c5b40a3..1df6328e7a 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -52,5 +52,6 @@ export const migrations = ( import("./migration/20260709013000_generic_session_input"), import("./migration/20260709025533_drop-todo"), import("./migration/20260709163752_time_suspended"), + import("./migration/20260709190621_session_pending_table"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260709190621_session_pending_table.ts b/packages/core/src/database/migration/20260709190621_session_pending_table.ts new file mode 100644 index 0000000000..1fc97a2257 --- /dev/null +++ b/packages/core/src/database/migration/20260709190621_session_pending_table.ts @@ -0,0 +1,35 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260709190621_session_pending_table", + up(tx) { + return Effect.gen(function* () { + // Beta reset: session_input becomes the pending-only session_pending + // table. Dropping the old table discards consumed ledger rows and any + // in-flight pending work along with every historical index variant. + yield* tx.run(`DROP TABLE \`session_input\`;`) + yield* tx.run(` + CREATE TABLE \`session_pending\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text NOT NULL, + \`type\` text NOT NULL, + \`data\` text NOT NULL, + \`delivery\` text, + \`admitted_seq\` integer NOT NULL, + \`time_created\` integer NOT NULL, + CONSTRAINT \`fk_session_pending_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE + ); + `) + yield* tx.run( + `CREATE INDEX \`session_pending_session_delivery_seq_idx\` ON \`session_pending\` (\`session_id\`,\`delivery\`,\`admitted_seq\`);`, + ) + yield* tx.run( + `CREATE UNIQUE INDEX \`session_pending_session_compaction_idx\` ON \`session_pending\` (\`session_id\`) WHERE "session_pending"."type" = 'compaction';`, + ) + yield* tx.run( + `CREATE UNIQUE INDEX \`session_pending_session_admitted_seq_idx\` ON \`session_pending\` (\`session_id\`,\`admitted_seq\`);`, + ) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 27651d0c13..f9b27c0e1a 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -166,19 +166,6 @@ export default { CONSTRAINT \`fk_part_message_id_message_id_fk\` FOREIGN KEY (\`message_id\`) REFERENCES \`message\`(\`id\`) ON DELETE CASCADE ); `) - yield* tx.run(` - CREATE TABLE \`session_input\` ( - \`id\` text PRIMARY KEY, - \`session_id\` text NOT NULL, - \`type\` text NOT NULL, - \`data\` text NOT NULL, - \`delivery\` text, - \`admitted_seq\` integer NOT NULL, - \`promoted_seq\` integer, - \`time_created\` integer NOT NULL, - CONSTRAINT \`fk_session_input_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE - ); - `) yield* tx.run(` CREATE TABLE \`session_message\` ( \`id\` text PRIMARY KEY, @@ -191,6 +178,18 @@ export default { CONSTRAINT \`fk_session_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE ); `) + yield* tx.run(` + CREATE TABLE \`session_pending\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text NOT NULL, + \`type\` text NOT NULL, + \`data\` text NOT NULL, + \`delivery\` text, + \`admitted_seq\` integer NOT NULL, + \`time_created\` integer NOT NULL, + CONSTRAINT \`fk_session_pending_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE + ); + `) yield* tx.run(` CREATE TABLE \`session\` ( \`id\` text PRIMARY KEY, @@ -249,18 +248,6 @@ export default { ) yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`) yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`) - yield* tx.run( - `CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`, - ) - yield* tx.run( - `CREATE UNIQUE INDEX \`session_input_session_pending_compaction_idx\` ON \`session_input\` (\`session_id\`) WHERE "session_input"."type" = 'compaction' and "session_input"."promoted_seq" is null;`, - ) - yield* tx.run( - `CREATE UNIQUE INDEX \`session_input_session_admitted_seq_idx\` ON \`session_input\` (\`session_id\`,\`admitted_seq\`);`, - ) - yield* tx.run( - `CREATE UNIQUE INDEX \`session_input_session_promoted_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`);`, - ) yield* tx.run( `CREATE UNIQUE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`, ) @@ -271,6 +258,15 @@ export default { `CREATE INDEX \`session_message_session_time_created_id_idx\` ON \`session_message\` (\`session_id\`,\`time_created\`,\`id\`);`, ) yield* tx.run(`CREATE INDEX \`session_message_time_created_idx\` ON \`session_message\` (\`time_created\`);`) + yield* tx.run( + `CREATE INDEX \`session_pending_session_delivery_seq_idx\` ON \`session_pending\` (\`session_id\`,\`delivery\`,\`admitted_seq\`);`, + ) + yield* tx.run( + `CREATE UNIQUE INDEX \`session_pending_session_compaction_idx\` ON \`session_pending\` (\`session_id\`) WHERE "session_pending"."type" = 'compaction';`, + ) + yield* tx.run( + `CREATE UNIQUE INDEX \`session_pending_session_admitted_seq_idx\` ON \`session_pending\` (\`session_id\`,\`admitted_seq\`);`, + ) yield* tx.run(`CREATE INDEX \`session_project_idx\` ON \`session\` (\`project_id\`);`) yield* tx.run(`CREATE INDEX \`session_workspace_idx\` ON \`session\` (\`workspace_id\`);`) yield* tx.run(`CREATE INDEX \`session_parent_idx\` ON \`session\` (\`parent_id\`);`) diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 3d3a3a5e45..96a7abafab 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -32,7 +32,7 @@ import { makeGlobalNode } from "./effect/app-node" import { LocationServiceMap } from "./location-service-map" import { MessageDecodeError } from "./session/error" import { SessionEvent } from "./session/event" -import { SessionInput } from "./session/input" +import { SessionPending } from "./session/pending" import { Snapshot } from "./snapshot" import { SessionRevert } from "./session/revert" import { Session } from "@opencode-ai/schema/session" @@ -187,6 +187,12 @@ export interface Interface { readonly context: ( sessionID: SessionSchema.ID, ) => Effect.Effect + /** + * Durable admitted session work not yet visible in projected history, + * ordered by admission. Includes unpromoted user and synthetic inputs and + * unhandled compaction barriers. + */ + readonly pending: (sessionID: SessionSchema.ID) => Effect.Effect /** * Durable, ordered, gap-free session log read. Replays public durable * session events after the exclusive `after` cursor, emits a `Synced` @@ -216,9 +222,9 @@ export interface Interface { files?: PromptInput.Prompt["files"] agents?: PromptInput.Prompt["agents"] metadata?: Record - delivery?: SessionInput.Delivery + delivery?: SessionPending.Delivery resume?: boolean - }) => Effect.Effect + }) => Effect.Effect readonly command: (input: { id?: SessionMessage.ID sessionID: SessionSchema.ID @@ -228,10 +234,10 @@ export interface Interface { model?: ModelV2.Ref files?: PromptInput.Prompt["files"] agents?: PromptInput.Prompt["agents"] - delivery?: SessionInput.Delivery + delivery?: SessionPending.Delivery resume?: boolean }) => Effect.Effect< - SessionInput.User, + SessionPending.User, NotFoundError | PromptConflictError | AttachmentError | CommandV2.NotFoundError | CommandV2.EvaluationError > readonly shell: (input: { @@ -247,7 +253,7 @@ export interface Interface { }) => Effect.Effect readonly compact: ( input: CompactInput, - ) => Effect.Effect + ) => Effect.Effect readonly wait: (id: SessionSchema.ID) => Effect.Effect readonly active: Effect.Effect> readonly background: (sessionID: SessionSchema.ID) => Effect.Effect @@ -259,9 +265,9 @@ export interface Interface { text: string description?: string metadata?: Record - delivery?: SessionInput.Delivery + delivery?: SessionPending.Delivery resume?: boolean - }) => Effect.Effect + }) => Effect.Effect readonly revert: { readonly stage: (input: { sessionID: SessionSchema.ID @@ -481,6 +487,10 @@ const layer = Layer.effect( yield* result.get(sessionID) return yield* store.context(sessionID) }), + pending: Effect.fn("V2Session.pending")(function* (sessionID) { + yield* result.get(sessionID) + return yield* SessionPending.list(db, sessionID) + }), log: (input) => Stream.unwrap( result @@ -504,25 +514,25 @@ const layer = Layer.effect( Effect.provideService(FSUtil.Service, fs), ) const messageID = input.id ?? SessionMessage.ID.create() - const admittedInput = SessionInput.Message.make({ + const admittedInput = SessionPending.Message.make({ type: "user", data: { ...prompt, metadata: input.metadata }, delivery: input.delivery ?? "steer", }) - const admitted = yield* SessionInput.admit(db, events, { + const admitted = yield* SessionPending.admit(db, events, { id: messageID, sessionID: input.sessionID, input: admittedInput, }).pipe( Effect.catchDefect((defect) => - defect instanceof SessionInput.LifecycleConflict + defect instanceof SessionPending.LifecycleConflict ? new PromptConflictError({ sessionID: input.sessionID, messageID }) : Effect.die(defect), ), ) if ( admitted.type !== "user" || - !SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput }) + !SessionPending.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput }) ) return yield* new PromptConflictError({ sessionID: input.sessionID, messageID }) if (input.resume !== false) { @@ -664,12 +674,12 @@ const layer = Layer.effect( compact: Effect.fn("V2Session.compact")(function* (input) { yield* result.get(input.sessionID) const inputID = input.id ?? SessionMessage.ID.create() - const admitted = yield* SessionInput.admitCompaction(db, events, { + const admitted = yield* SessionPending.admitCompaction(db, events, { id: inputID, sessionID: input.sessionID, }).pipe( Effect.catchDefect((defect) => - defect instanceof SessionInput.LifecycleConflict + defect instanceof SessionPending.LifecycleConflict ? new CompactionConflictError({ sessionID: input.sessionID, inputID }) : Effect.die(defect), ), @@ -709,7 +719,7 @@ const layer = Layer.effect( Effect.gen(function* () { yield* result.get(input.sessionID) const inputID = input.id ?? SessionMessage.ID.create() - const admittedInput = SessionInput.Message.make({ + const admittedInput = SessionPending.Message.make({ type: "synthetic", data: { text: input.text, @@ -718,20 +728,20 @@ const layer = Layer.effect( }, delivery: input.delivery ?? "steer", }) - const admitted = yield* SessionInput.admit(db, events, { + const admitted = yield* SessionPending.admit(db, events, { id: inputID, sessionID: input.sessionID, input: admittedInput, }).pipe( Effect.catchDefect((defect) => - defect instanceof SessionInput.LifecycleConflict + defect instanceof SessionPending.LifecycleConflict ? new SyntheticConflictError({ sessionID: input.sessionID, inputID }) : Effect.die(defect), ), ) if ( admitted.type !== "synthetic" || - !SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput }) + !SessionPending.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput }) ) return yield* new SyntheticConflictError({ sessionID: input.sessionID, inputID }) if (input.resume !== false && !(yield* result.get(input.sessionID)).revert) diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/pending.ts similarity index 58% rename from packages/core/src/session/input.ts rename to packages/core/src/session/pending.ts index e3e62ac139..70f0b34c5b 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/pending.ts @@ -1,6 +1,6 @@ -export * as SessionInput from "./input" +export * as SessionPending from "./pending" -import { and, asc, eq, isNull } from "drizzle-orm" +import { and, asc, eq } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" import { Compaction, @@ -11,14 +11,16 @@ import { SyntheticData, User, UserData, -} from "@opencode-ai/schema/session-input" +} from "@opencode-ai/schema/session-pending" +import { Event } from "@opencode-ai/schema/event" import type { Database } from "../database/database" import type { EventV2 } from "../event" +import { EventTable } from "../event/sql" import { KeyedMutex } from "../effect/keyed-mutex" import { SessionEvent } from "./event" import { SessionMessage } from "./message" import { SessionSchema } from "./schema" -import { SessionInputTable, SessionMessageTable } from "./sql" +import { SessionMessageTable, SessionPendingTable } from "./sql" type DatabaseService = Database.Interface["db"] @@ -28,25 +30,28 @@ const decodeUser = Schema.decodeUnknownSync(UserData) const encodeUser = Schema.encodeSync(UserData) const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData) const encodeSynthetic = Schema.encodeSync(SyntheticData) +const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data) +const admittedEventType = Event.versionedType( + SessionEvent.InputAdmitted.type, + SessionEvent.InputAdmitted.durable.version, +) const inboxLocks = KeyedMutex.makeUnsafe() -export class LifecycleConflict extends Schema.TaggedErrorClass()("SessionInput.LifecycleConflict", { - id: SessionMessage.ID, -}) {} +export class LifecycleConflict extends Schema.TaggedErrorClass()( + "SessionPending.LifecycleConflict", + { + id: SessionMessage.ID, + }, +) {} -const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => { +const fromRow = (row: typeof SessionPendingTable.$inferSelect): Info => { const base = { admittedSeq: row.admitted_seq, id: SessionMessage.ID.make(row.id), sessionID: SessionSchema.ID.make(row.session_id), timeCreated: DateTime.makeUnsafe(row.time_created), } - if (row.type === "compaction") - return Compaction.make({ - ...base, - type: "compaction", - ...(row.promoted_seq === null ? {} : { handledSeq: row.promoted_seq }), - }) + if (row.type === "compaction") return Compaction.make({ ...base, type: "compaction" }) if (!row.delivery) throw new LifecycleConflict({ id: base.id }) if (row.type === "user") return User.make({ @@ -54,7 +59,6 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => { type: "user", data: decodeUser(row.data), delivery: row.delivery, - ...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }), }) if (row.type === "synthetic") return Synthetic.make({ @@ -62,31 +66,29 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => { type: "synthetic", data: decodeSynthetic(row.data), delivery: row.delivery, - ...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }), }) throw new LifecycleConflict({ id: base.id }) } -export const find = Effect.fn("SessionInput.find")(function* (db: DatabaseService, id: SessionMessage.ID) { - const row = yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie) +export const find = Effect.fn("SessionPending.find")(function* (db: DatabaseService, id: SessionMessage.ID) { + const row = yield* db + .select() + .from(SessionPendingTable) + .where(eq(SessionPendingTable.id, id)) + .get() + .pipe(Effect.orDie) return row === undefined ? undefined : fromRow(row) }) -export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(function* ( +export const compaction = Effect.fn("SessionPending.compaction")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, ) { const row = yield* db .select() - .from(SessionInputTable) - .where( - and( - eq(SessionInputTable.session_id, sessionID), - eq(SessionInputTable.type, "compaction"), - isNull(SessionInputTable.promoted_seq), - ), - ) - .orderBy(asc(SessionInputTable.admitted_seq)) + .from(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.type, "compaction"))) + .orderBy(asc(SessionPendingTable.admitted_seq)) .limit(1) .get() .pipe(Effect.orDie) @@ -95,7 +97,51 @@ export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(fun return entry.type === "compaction" ? entry : undefined }) -export const admit = Effect.fn("SessionInput.admit")(function* ( +/** + * Reconstruct the admitted record for a pending row that was already consumed + * by promotion. The projected `session_message` row proves promotion happened; + * the durable `session.input.admitted` event retains the exact admitted + * message, including delivery. + */ +const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + id: SessionMessage.ID, +) { + const message = yield* db + .select() + .from(SessionMessageTable) + .where(eq(SessionMessageTable.id, id)) + .get() + .pipe(Effect.orDie) + if (message === undefined) return undefined + if (message.session_id !== sessionID || (message.type !== "user" && message.type !== "synthetic")) + return yield* Effect.die(new LifecycleConflict({ id })) + const rows = yield* db + .select() + .from(EventTable) + .where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, admittedEventType))) + .all() + .pipe(Effect.orDie) + for (const row of rows) { + const decoded = decodeAdmittedEvent(row.data) + if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue + const base = { + admittedSeq: row.seq, + id, + sessionID, + timeCreated: DateTime.makeUnsafe(row.created), + } + return decoded.value.input.type === "user" + ? User.make({ ...base, ...decoded.value.input }) + : Synthetic.make({ ...base, ...decoded.value.input }) + } + // A projected message without an admitted event in this aggregate (for + // example fork-copied history) is not a retryable admission. + return yield* Effect.die(new LifecycleConflict({ id })) +}) + +export const admit = Effect.fn("SessionPending.admit")(function* ( db: DatabaseService, events: EventV2.Interface, request: { @@ -109,6 +155,8 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id })) return existing } + const promoted = yield* promotedFromHistory(db, request.sessionID, request.id) + if (promoted !== undefined) return promoted return yield* events .publish(SessionEvent.InputAdmitted, { inputID: request.id, @@ -141,7 +189,7 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( ) }) -export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(function* ( +export const admitCompaction = Effect.fn("SessionPending.admitCompaction")(function* ( db: DatabaseService, events: EventV2.Interface, input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID }, @@ -153,7 +201,7 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact return yield* Effect.die(new LifecycleConflict({ id: input.id })) } - const pending = yield* pendingCompaction(db, input.sessionID) + const pending = yield* compaction(db, input.sessionID) if (pending) return pending return yield* events .publish(SessionEvent.Compaction.Admitted, { @@ -164,14 +212,14 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio Effect.flatMap((event) => { if (event.durable === undefined) return Effect.die(new Error("Compaction admission event is missing aggregate sequence")) - return pendingCompaction(db, input.sessionID).pipe( + return compaction(db, input.sessionID).pipe( Effect.flatMap((stored) => stored ? Effect.succeed(stored) : Effect.die(new LifecycleConflict({ id: input.id })), ), ) }), Effect.catchDefect((defect) => - pendingCompaction(db, input.sessionID).pipe( + compaction(db, input.sessionID).pipe( Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))), ), ), @@ -180,7 +228,7 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio ) }) -export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(function* ( +export const projectAdmitted = Effect.fn("SessionPending.projectAdmitted")(function* ( db: DatabaseService, request: { readonly admittedSeq: number @@ -198,7 +246,7 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio .pipe(Effect.orDie) if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: request.id })) const stored = yield* db - .insert(SessionInputTable) + .insert(SessionPendingTable) .values({ id: request.id, session_id: request.sessionID, @@ -209,13 +257,13 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio time_created: DateTime.toEpochMillis(request.timeCreated), }) .onConflictDoNothing() - .returning({ id: SessionInputTable.id }) + .returning({ id: SessionPendingTable.id }) .get() .pipe(Effect.orDie) if (!stored) return yield* Effect.die(new LifecycleConflict({ id: request.id })) }) -export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompactionAdmitted")(function* ( +export const projectCompactionAdmitted = Effect.fn("SessionPending.projectCompactionAdmitted")(function* ( db: DatabaseService, input: { readonly admittedSeq: number @@ -232,7 +280,7 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti .pipe(Effect.orDie) if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: input.id })) const stored = yield* db - .insert(SessionInputTable) + .insert(SessionPendingTable) .values({ id: input.id, session_id: input.sessionID, @@ -249,84 +297,74 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti const entry = fromRow(stored) return entry.type === "compaction" ? entry : yield* Effect.die(new LifecycleConflict({ id: entry.id })) } - const pending = yield* pendingCompaction(db, input.sessionID) + const pending = yield* compaction(db, input.sessionID) if (pending) return pending return yield* Effect.die(new LifecycleConflict({ id: input.id })) }) -export const projectPromoted = Effect.fn("SessionInput.projectPromoted")(function* ( +/** + * Consume one pending row at promotion. The row's content feeds the projected + * message insert inside the same event transaction; the deleted row is what + * makes the table pending-only. + */ +export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(function* ( db: DatabaseService, input: { readonly id: SessionMessage.ID readonly sessionID: SessionSchema.ID - readonly promotedSeq: number }, ) { - if (yield* pendingCompaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id })) - const updated = yield* db - .update(SessionInputTable) - .set({ promoted_seq: input.promotedSeq }) - .where( - and( - eq(SessionInputTable.id, input.id), - eq(SessionInputTable.session_id, input.sessionID), - isNull(SessionInputTable.promoted_seq), - ), - ) + if (yield* compaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id })) + const deleted = yield* db + .delete(SessionPendingTable) + .where(and(eq(SessionPendingTable.id, input.id), eq(SessionPendingTable.session_id, input.sessionID))) .returning() .get() .pipe(Effect.orDie) - const stored = updated ? fromRow(updated) : yield* find(db, input.id) - if ( - !stored || - stored.type === "compaction" || - stored.sessionID !== input.sessionID || - stored.promotedSeq !== input.promotedSeq - ) - return yield* Effect.die(new LifecycleConflict({ id: input.id })) + if (!deleted) return yield* Effect.die(new LifecycleConflict({ id: input.id })) + const stored = fromRow(deleted) + if (stored.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: input.id })) return stored }) -export const settleCompaction = Effect.fn("SessionInput.settleCompaction")(function* ( +export const settleCompaction = Effect.fn("SessionPending.settleCompaction")(function* ( db: DatabaseService, - input: { readonly sessionID: SessionSchema.ID; readonly handledSeq: number }, + input: { readonly sessionID: SessionSchema.ID }, ) { - const updated = yield* db - .update(SessionInputTable) - .set({ promoted_seq: input.handledSeq }) - .where( - and( - eq(SessionInputTable.session_id, input.sessionID), - eq(SessionInputTable.type, "compaction"), - isNull(SessionInputTable.promoted_seq), - ), - ) + const deleted = yield* db + .delete(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, input.sessionID), eq(SessionPendingTable.type, "compaction"))) .returning() .get() .pipe(Effect.orDie) - if (updated) { - const stored = fromRow(updated) + if (deleted) { + const stored = fromRow(deleted) return stored.type === "compaction" ? stored : yield* Effect.die(new LifecycleConflict({ id: stored.id })) } return undefined }) -export const hasPending = Effect.fn("SessionInput.hasPending")(function* ( +export const list = Effect.fn("SessionPending.list")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { + const rows = yield* db + .select() + .from(SessionPendingTable) + .where(eq(SessionPendingTable.session_id, sessionID)) + .orderBy(asc(SessionPendingTable.admitted_seq)) + .all() + .pipe(Effect.orDie) + return rows.map(fromRow) +}) + +export const has = Effect.fn("SessionPending.has")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, delivery: Delivery, ) { - if (yield* pendingCompaction(db, sessionID)) return false + if (yield* compaction(db, sessionID)) return false const row = yield* db - .select({ id: SessionInputTable.id }) - .from(SessionInputTable) - .where( - and( - eq(SessionInputTable.session_id, sessionID), - isNull(SessionInputTable.promoted_seq), - eq(SessionInputTable.delivery, delivery), - ), - ) + .select({ id: SessionPendingTable.id }) + .from(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, delivery))) .limit(1) .get() .pipe(Effect.orDie) @@ -350,15 +388,15 @@ export const equivalent = ( return false } -const publish = Effect.fn("SessionInput.publish")(function* ( +const publish = Effect.fn("SessionPending.publish")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, - rows: ReadonlyArray, + rows: ReadonlyArray, ) { return yield* inboxLocks.withLock(sessionID)( Effect.gen(function* () { - if (yield* pendingCompaction(db, sessionID)) return 0 + if (yield* compaction(db, sessionID)) return 0 yield* Effect.forEach( rows, (row) => { @@ -372,12 +410,8 @@ const publish = Effect.fn("SessionInput.publish")(function* ( .pipe( Effect.catchDefect((defect) => defect instanceof LifecycleConflict - ? find(db, entry.id).pipe( - Effect.flatMap((stored) => - stored?.type !== "compaction" && stored?.promotedSeq !== undefined - ? Effect.void - : Effect.die(defect), - ), + ? promotedFromHistory(db, sessionID, entry.id).pipe( + Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))), ) : Effect.die(defect), ), @@ -390,45 +424,33 @@ const publish = Effect.fn("SessionInput.publish")(function* ( ) }) -export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( +export const promoteSteers = Effect.fn("SessionPending.promoteSteers")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, ) { - if (yield* pendingCompaction(db, sessionID)) return 0 + if (yield* compaction(db, sessionID)) return 0 const rows = yield* db .select() - .from(SessionInputTable) - .where( - and( - eq(SessionInputTable.session_id, sessionID), - isNull(SessionInputTable.promoted_seq), - eq(SessionInputTable.delivery, "steer"), - ), - ) - .orderBy(asc(SessionInputTable.admitted_seq)) + .from(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "steer"))) + .orderBy(asc(SessionPendingTable.admitted_seq)) .all() .pipe(Effect.orDie) return yield* publish(db, events, sessionID, rows) }) -export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(function* ( +export const promoteNextQueued = Effect.fn("SessionPending.promoteNextQueued")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, ) { - if (yield* pendingCompaction(db, sessionID)) return false + if (yield* compaction(db, sessionID)) return false const row = yield* db .select() - .from(SessionInputTable) - .where( - and( - eq(SessionInputTable.session_id, sessionID), - isNull(SessionInputTable.promoted_seq), - eq(SessionInputTable.delivery, "queue"), - ), - ) - .orderBy(asc(SessionInputTable.admitted_seq)) + .from(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "queue"))) + .orderBy(asc(SessionPendingTable.admitted_seq)) .limit(1) .get() .pipe(Effect.orDie) diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index a3c2b53ae4..4d302108b5 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -1,6 +1,6 @@ export * as SessionProjector from "./projector" -import { and, asc, desc, eq, gt, gte, inArray, lt, or, sql } from "drizzle-orm" +import { and, asc, desc, eq, gt, gte, inArray, lt, sql } from "drizzle-orm" import { DateTime, Effect, Layer, Schema, Stream } from "effect" import { Database } from "../database/database" import { EventV2 } from "../event" @@ -11,14 +11,14 @@ import { SessionV1 } from "../v1/session" import { WorkspaceTable } from "../control-plane/workspace.sql" import { SessionMessage } from "./message" import { SessionMessageUpdater } from "./message-updater" -import { SessionInput } from "./input" +import { SessionPending } from "./pending" import { WorkspaceV2 } from "../workspace" import { InstructionCheckpoint } from "./instruction-checkpoint" import { MessageTable, PartTable, InstructionCheckpointTable, - SessionInputTable, + SessionPendingTable, SessionMessageTable, SessionTable, } from "./sql" @@ -293,25 +293,25 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( .run() .pipe(Effect.orDie) - const inputRows = yield* db + const pendingRows = yield* db .select() - .from(SessionInputTable) + .from(SessionPendingTable) .where( and( - eq(SessionInputTable.session_id, event.data.parentID), + eq(SessionPendingTable.session_id, event.data.parentID), inArray( - SessionInputTable.id, + SessionPendingTable.id, rows.map((row) => row.id), ), ), ) .all() .pipe(Effect.orDie) - if (inputRows.length > 0) { + if (pendingRows.length > 0) { yield* db - .insert(SessionInputTable) + .insert(SessionPendingTable) .values( - inputRows.flatMap((row) => { + pendingRows.flatMap((row) => { const id = idMap.get(row.id) return id && row.type !== "compaction" ? [ @@ -322,7 +322,6 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( data: row.data, delivery: row.delivery, admitted_seq: row.admitted_seq, - promoted_seq: row.promoted_seq, time_created: row.time_created, }, ] @@ -633,10 +632,9 @@ const layer = Layer.effectDiscard( Effect.gen(function* () { if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) - const input = yield* SessionInput.projectPromoted(db, { + const input = yield* SessionPending.projectPromoted(db, { id: event.data.inputID, sessionID: event.data.sessionID, - promotedSeq: event.durable.seq, }) yield* insertMessage( db, @@ -666,7 +664,7 @@ const layer = Layer.effectDiscard( Effect.gen(function* () { if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) - yield* SessionInput.projectAdmitted(db, { + yield* SessionPending.projectAdmitted(db, { admittedSeq: event.durable.seq, id: event.data.inputID, sessionID: event.data.sessionID, @@ -679,7 +677,7 @@ const layer = Layer.effectDiscard( Effect.gen(function* () { if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) - yield* SessionInput.projectCompactionAdmitted(db, { + yield* SessionPending.projectCompactionAdmitted(db, { admittedSeq: event.durable.seq, id: event.data.inputID, sessionID: event.data.sessionID, @@ -727,10 +725,7 @@ const layer = Layer.effectDiscard( if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) if (event.data.reason === "manual") - yield* SessionInput.settleCompaction(db, { - sessionID: event.data.sessionID, - handledSeq: event.durable.seq, - }) + yield* SessionPending.settleCompaction(db, { sessionID: event.data.sessionID }) }), ) yield* events.project(SessionEvent.Compaction.Failed, (event) => @@ -739,10 +734,7 @@ const layer = Layer.effectDiscard( if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) if (event.data.reason === "manual") - yield* SessionInput.settleCompaction(db, { - sessionID: event.data.sessionID, - handledSeq: event.durable.seq, - }) + yield* SessionPending.settleCompaction(db, { sessionID: event.data.sessionID }) }), ) yield* events.project(SessionEvent.RevertEvent.Staged, (event) => @@ -786,11 +778,11 @@ const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie) yield* db - .delete(SessionInputTable) + .delete(SessionPendingTable) .where( and( - eq(SessionInputTable.session_id, event.data.sessionID), - or(gte(SessionInputTable.admitted_seq, boundary.seq), gte(SessionInputTable.promoted_seq, boundary.seq)), + eq(SessionPendingTable.session_id, event.data.sessionID), + gte(SessionPendingTable.admitted_seq, boundary.seq), ), ) .run() diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 7570585fc1..6f46931332 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -34,7 +34,7 @@ import { InstructionCheckpoint } from "../instruction-checkpoint" import { SessionCompaction } from "../compaction" import { SessionEvent } from "../event" import { SessionHistory } from "../history" -import { SessionInput } from "../input" +import { SessionPending } from "../pending" import { SessionMessage } from "../message" import { SessionSchema } from "../schema" import { SessionStore } from "../store" @@ -203,7 +203,7 @@ const layer = Layer.effect( const attemptStep = Effect.fn("SessionRunner.attemptStep")(function* ( sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, + promotion: SessionPending.Delivery | undefined, step: number, recoverOverflow?: typeof compaction.compactAfterOverflow, assistantMessageID?: SessionMessage.ID, @@ -226,10 +226,10 @@ const layer = Layer.effect( let currentStep = step if (promotion) { let promoted = 0 - if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id) + if (promotion === "steer") promoted = yield* SessionPending.promoteSteers(db, events, session.id) if (promotion === "queue") { - promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id)) - promoted += yield* SessionInput.promoteSteers(db, events, session.id) + promoted += Number(yield* SessionPending.promoteNextQueued(db, events, session.id)) + promoted += yield* SessionPending.promoteSteers(db, events, session.id) } if (promoted > 0) currentStep = 1 } @@ -285,7 +285,7 @@ const layer = Layer.effect( const advertisedTools = new Set(hookedRequest.tools.map((tool) => tool.name)) // Automatic compaction completed; rebuild the request from compacted history. if ( - !(yield* SessionInput.pendingCompaction(db, session.id)) && + !(yield* SessionPending.compaction(db, session.id)) && (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request: hookedRequest })) ) return { _tag: "RestartAfterCompaction", step: currentStep } as const @@ -550,7 +550,7 @@ const layer = Layer.effect( const runStep = Effect.fnUntraced(function* ( sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, + promotion: SessionPending.Delivery | undefined, step: number, ) { // Compaction restarts rebuild the request from compacted history without re-promoting. @@ -595,7 +595,7 @@ const layer = Layer.effect( const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( sessionID: SessionSchema.ID, ) { - const pending = yield* SessionInput.pendingCompaction(db, sessionID) + const pending = yield* SessionPending.compaction(db, sessionID) if (!pending) return false const session = yield* getSession(sessionID) return yield* Effect.uninterruptibleMask((restore) => @@ -611,7 +611,7 @@ const layer = Layer.effect( ).pipe(Effect.exit) if (Exit.isSuccess(compacted) && compacted.value) return true if (Exit.isFailure(compacted)) { - const unsettled = yield* SessionInput.pendingCompaction(db, sessionID) + const unsettled = yield* SessionPending.compaction(db, sessionID) if (unsettled) yield* events.publish(SessionEvent.Compaction.Failed, { sessionID, @@ -621,7 +621,7 @@ const layer = Layer.effect( }) return yield* Effect.failCause(compacted.cause) } - const unsettled = yield* SessionInput.pendingCompaction(db, sessionID) + const unsettled = yield* SessionPending.compaction(db, sessionID) if (unsettled) yield* events.publish(SessionEvent.Compaction.Failed, { sessionID, @@ -640,11 +640,11 @@ const layer = Layer.effect( readonly force: boolean }) { yield* runPendingCompaction(input.sessionID) - const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") - const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") + const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer") + const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue") if (!input.force && !hasSteer && !hasQueue) return yield* failInterruptedTools(input.sessionID) - let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined + let promotion: SessionPending.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined let shouldRun = input.force || hasSteer || hasQueue while (shouldRun) { let needsContinuation = true @@ -664,16 +664,16 @@ const layer = Layer.effect( needsContinuation = result.needsContinuation step = result.step + 1 if (needsContinuation) { - promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer" + promotion = (yield* SessionPending.compaction(db, input.sessionID)) ? undefined : "steer" continue } yield* runPendingCompaction(input.sessionID) promotion = "steer" - needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") + needsContinuation = yield* SessionPending.has(db, input.sessionID, "steer") } yield* runPendingCompaction(input.sessionID) - const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") - const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") + const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer") + const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue") shouldRun = hasSteer || hasQueue promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined } diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index f5e97aad76..505ecec383 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -3,7 +3,7 @@ import { sql } from "drizzle-orm" import { directoryColumn, pathColumn } from "../database/path" import { ProjectTable } from "../project/sql" import type { SessionMessage } from "./message" -import type { SessionInput } from "./input" +import type { SessionPending } from "./pending" import type { FileDiff } from "@opencode-ai/schema/file-diff" import { PermissionV1 } from "../v1/permission" import { ProjectV2 } from "../project" @@ -13,7 +13,7 @@ import { WorkspaceV2 } from "../workspace" import { Timestamps } from "../database/schema.sql" import type { Instructions } from "../instructions/index" import type { Session } from "@opencode-ai/schema/session" -import type { SyntheticData, UserData } from "@opencode-ai/schema/session-input" +import type { SyntheticData, UserData } from "@opencode-ai/schema/session-pending" import type { RevertV1 } from "@opencode-ai/schema/session-revert" import type { Schema } from "effect" @@ -126,35 +126,28 @@ export const SessionMessageTable = sqliteTable( ], ) -export const SessionInputTable = sqliteTable( - "session_input", +export const SessionPendingTable = sqliteTable( + "session_pending", { id: text().$type().primaryKey(), session_id: text() .$type() .notNull() .references(() => SessionTable.id, { onDelete: "cascade" }), - type: text().$type().notNull(), + type: text().$type().notNull(), data: text({ mode: "json" }).$type>().notNull(), - delivery: text().$type(), + delivery: text().$type(), admitted_seq: integer().notNull(), - promoted_seq: integer(), time_created: integer() .notNull() .$default(() => Date.now()), }, (table) => [ - index("session_input_session_pending_delivery_seq_idx").on( - table.session_id, - table.promoted_seq, - table.delivery, - table.admitted_seq, - ), - uniqueIndex("session_input_session_pending_compaction_idx") + index("session_pending_session_delivery_seq_idx").on(table.session_id, table.delivery, table.admitted_seq), + uniqueIndex("session_pending_session_compaction_idx") .on(table.session_id) - .where(sql`${table.type} = 'compaction' and ${table.promoted_seq} is null`), - uniqueIndex("session_input_session_admitted_seq_idx").on(table.session_id, table.admitted_seq), - uniqueIndex("session_input_session_promoted_seq_idx").on(table.session_id, table.promoted_seq), + .where(sql`${table.type} = 'compaction'`), + uniqueIndex("session_pending_session_admitted_seq_idx").on(table.session_id, table.admitted_seq), ], ) diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index 22a13a1c04..436f680cc3 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -11,14 +11,15 @@ import { migrations } from "@opencode-ai/core/database/migration.gen" import sessionUsageMigration from "@opencode-ai/core/database/migration/20260510033149_session_usage" import normalizeStoragePathsMigration from "@opencode-ai/core/database/migration/20260601010001_normalize_storage_paths" import sessionMessageProjectionOrderMigration from "@opencode-ai/core/database/migration/20260603040000_session_message_projection_order" -import eventSourcedSessionInputMigration from "@opencode-ai/core/database/migration/20260604172448_event_sourced_session_input" +import eventSourcedSessionPendingMigration from "@opencode-ai/core/database/migration/20260604172448_event_sourced_session_input" import contextEpochAgentMigration from "@opencode-ai/core/database/migration/20260605042240_add_context_epoch_agent" import simplifyIntegrationCredentialsMigration from "@opencode-ai/core/database/migration/20260611192811_lush_chimera" -import simplifySessionInputMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input" +import simplifySessionPendingMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input" import resetSessionEventsMigration from "@opencode-ai/core/database/migration/20260703200000_reset_v2_session_events" import durableSessionInboxMigration from "@opencode-ai/core/database/migration/20260707010146_durable_session_inbox" import migratePrelaunchV2StateMigration from "@opencode-ai/core/database/migration/20260707120000_migrate_prelaunch_v2_state" -import genericSessionInputMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input" +import genericSessionPendingMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input" +import sessionPendingTableMigration from "@opencode-ai/core/database/migration/20260709190621_session_pending_table" import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions" import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork" import timeSuspendedMigration from "@opencode-ai/core/database/migration/20260709163752_time_suspended" @@ -352,7 +353,10 @@ describe("DatabaseMigration", () => { }) expect( yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`), - ).toEqual({ name: "session_input" }) + ).toBeUndefined() + expect( + yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_pending'`), + ).toEqual({ name: "session_pending" }) expect( yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'instruction_checkpoint'`), ).toEqual({ name: "instruction_checkpoint" }) @@ -364,18 +368,17 @@ describe("DatabaseMigration", () => { expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: migrations.length }) expect( yield* db.all( - sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`, + sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_pending_session_delivery_seq_idx', 'session_pending_session_compaction_idx', 'session_pending_session_admitted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`, ), ).toEqual([ { name: "event_aggregate_seq_idx" }, { name: "event_aggregate_type_seq_idx" }, - { name: "session_input_session_admitted_seq_idx" }, - { name: "session_input_session_pending_compaction_idx" }, - { name: "session_input_session_pending_delivery_seq_idx" }, - { name: "session_input_session_promoted_seq_idx" }, { name: "session_message_session_seq_idx" }, { name: "session_message_session_time_created_id_idx" }, { name: "session_message_session_type_seq_idx" }, + { name: "session_pending_session_admitted_seq_idx" }, + { name: "session_pending_session_compaction_idx" }, + { name: "session_pending_session_delivery_seq_idx" }, ]) }), ) @@ -568,7 +571,7 @@ describe("DatabaseMigration", () => { sql`INSERT INTO session_input (id, session_id, prompt, delivery, time_created) VALUES ('msg_pending', 'session', '{}', 'steer', 1)`, ) - yield* DatabaseMigration.applyOnly(db, [eventSourcedSessionInputMigration]) + yield* DatabaseMigration.applyOnly(db, [eventSourcedSessionPendingMigration]) expect(yield* db.all(sql`SELECT id, workspace_id FROM session`)).toEqual([ { id: "session", workspace_id: null }, @@ -631,7 +634,7 @@ describe("DatabaseMigration", () => { sql`INSERT INTO event (id, aggregate_id, seq, type, data, created) VALUES ('event', 'session', 9, 'session.updated.1', '{}', 1)`, ) yield* db.run( - sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`, + sql`INSERT INTO session_pending (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`, ) yield* db.run( sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('projected', 'session', 'user', 9, 1, 1, '{}')`, @@ -640,9 +643,17 @@ describe("DatabaseMigration", () => { sql`INSERT INTO instruction_checkpoint (session_id, baseline, snapshot, baseline_seq) VALUES ('session', 'baseline', '{}', 9)`, ) yield* db.run(sql`ALTER TABLE instruction_checkpoint RENAME TO session_context_epoch`) - yield* db.run(sql`DELETE FROM migration WHERE id = ${simplifySessionInputMigration.id}`) - yield* DatabaseMigration.applyOnly(db, [simplifySessionInputMigration]) + // The partial compaction index embeds the qualified table name, so it + // must drop before the historical rename dance and recreate after. + yield* db.run(sql`DROP INDEX session_pending_session_compaction_idx`) + yield* db.run(sql`ALTER TABLE session_pending RENAME TO session_input`) + yield* db.run(sql`DELETE FROM migration WHERE id = ${simplifySessionPendingMigration.id}`) + yield* DatabaseMigration.applyOnly(db, [simplifySessionPendingMigration]) yield* db.run(sql`ALTER TABLE session_context_epoch RENAME TO instruction_checkpoint`) + yield* db.run(sql`ALTER TABLE session_input RENAME TO session_pending`) + yield* db.run( + sql`CREATE UNIQUE INDEX session_pending_session_compaction_idx ON session_pending (session_id) WHERE "session_pending"."type" = 'compaction'`, + ) const database = Layer.succeed(Database.Service, { db }) yield* EventV2.Service.use((service) => @@ -672,7 +683,7 @@ describe("DatabaseMigration", () => { (SELECT COUNT(*) FROM message WHERE id = 'message') AS messages, (SELECT COUNT(*) FROM part WHERE id = 'part') AS parts, (SELECT COUNT(*) FROM workspace) AS workspaces, - (SELECT COUNT(*) FROM session_input) AS sessionInputs, + (SELECT COUNT(*) FROM session_pending) AS sessionInputs, (SELECT COUNT(*) FROM session_message) AS sessionMessages, (SELECT COUNT(*) FROM instruction_checkpoint) AS instructionCheckpoints, (SELECT seq FROM event_sequence WHERE aggregate_id = 'session') AS seq, @@ -756,7 +767,7 @@ describe("DatabaseMigration", () => { sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-promoted', 'session', 7, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"empty"}')`, ) - yield* DatabaseMigration.applyOnly(db, [genericSessionInputMigration]) + yield* DatabaseMigration.applyOnly(db, [genericSessionPendingMigration]) expect(yield* db.all(sql`SELECT id, type, data, delivery FROM session_input ORDER BY admitted_seq`)).toEqual([ { id: "input", type: "user", data: '{"text":"hello"}', delivery: "queue" }, @@ -775,6 +786,47 @@ describe("DatabaseMigration", () => { ) }) + test("replaces the durable inbox with the empty session_pending table", async () => { + await run( + Effect.gen(function* () { + const db = yield* makeDb + yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`) + yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`) + yield* db.run( + sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL REFERENCES session(id) ON DELETE CASCADE, type text NOT NULL, data text NOT NULL, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`, + ) + // Interim v2 builds shipped differing index sets on real databases; + // dropping the table removes whatever variant exists. + yield* db.run( + sql`CREATE INDEX session_input_session_pending_type_delivery_seq_idx ON session_input (session_id, promoted_seq, type, delivery, admitted_seq)`, + ) + yield* db.run( + sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, promoted_seq, time_created) VALUES ('pending', 'session', 'user', '{"text":"hello"}', 'steer', 4, NULL, 1)`, + ) + + yield* DatabaseMigration.applyOnly(db, [sessionPendingTableMigration]) + + expect( + yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`), + ).toBeUndefined() + expect(yield* db.all(sql`SELECT id FROM session_pending`)).toEqual([]) + expect( + (yield* db.all<{ name: string }>(sql`PRAGMA table_info(session_pending)`)).map((column) => column.name), + ).toEqual(["id", "session_id", "type", "data", "delivery", "admitted_seq", "time_created"]) + expect( + (yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(session_pending)`)) + .filter((index) => index.name.startsWith("session_")) + .map((index) => ({ name: index.name, unique: index.unique })) + .sort((a, b) => a.name.localeCompare(b.name)), + ).toEqual([ + { name: "session_pending_session_admitted_seq_idx", unique: 1 }, + { name: "session_pending_session_compaction_idx", unique: 1 }, + { name: "session_pending_session_delivery_seq_idx", unique: 0 }, + ]) + }), + ) + }) + test("resets incompatible projected Session messages before adding sequence order", async () => { await run( Effect.gen(function* () { diff --git a/packages/core/test/session-compact.test.ts b/packages/core/test/session-compact.test.ts index 814e8c1872..f0e4049cc4 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -15,7 +15,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionV2 } from "@opencode-ai/core/session" import { SessionCompaction } from "@opencode-ai/core/session/compaction" import { SessionEvent } from "@opencode-ai/core/session/event" -import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionPending } from "@opencode-ai/core/session/pending" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" @@ -106,7 +106,7 @@ describe("SessionV2.compact", () => { expect(second.id).toBe(first.id) expect(requests).toHaveLength(0) - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, created.id)).toMatchObject({ + expect(yield* SessionPending.compaction((yield* Database.Service).db, created.id)).toMatchObject({ id: first.id, }) expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toBeUndefined() diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index e1b83b4f50..4b43fb5477 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -20,7 +20,7 @@ import { SessionV1 } from "@opencode-ai/core/v1/session" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" -import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionPending } from "@opencode-ai/core/session/pending" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" @@ -195,9 +195,9 @@ describe("SessionV2.create", () => { text: "First", resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id) + yield* SessionPending.promoteSteers(db, events, parent.id) yield* session.synthetic({ sessionID: parent.id, text: "parent note", resume: false }) - yield* SessionInput.promoteSteers(db, events, parent.id) + yield* SessionPending.promoteSteers(db, events, parent.id) const forked = yield* session.fork({ sessionID: parent.id }) const parentContext = yield* session.context(parent.id) @@ -217,30 +217,28 @@ describe("SessionV2.create", () => { durable: { seq: 0 }, data: { sessionID: forked.id, parentID: parent.id }, }) - expect(yield* SessionInput.find(db, forkContext[0].id)).toMatchObject({ - sessionID: forked.id, - type: "user", - data: { text: "First" }, - promotedSeq: 2, - }) - expect(yield* SessionInput.find(db, forkContext[1].id)).toMatchObject({ - sessionID: forked.id, - type: "synthetic", - data: { text: "parent note" }, - }) + expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined() + expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined() + // Fork-copied messages have no admitted event in the fork aggregate, so + // reusing their IDs as prompt IDs is conflicting reuse, not a retry. + expect( + yield* session + .prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }) + .pipe(Effect.flip), + ).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id }) yield* session.prompt({ sessionID: parent.id, text: "Parent changed", resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id) + yield* SessionPending.promoteSteers(db, events, parent.id) yield* session.prompt({ sessionID: forked.id, text: "Child continues", resume: false, }) - yield* SessionInput.promoteSteers(db, events, forked.id) + yield* SessionPending.promoteSteers(db, events, forked.id) expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) @@ -250,7 +248,7 @@ describe("SessionV2.create", () => { (event): number | undefined => event.durable?.seq, ), ).toEqual([0, 5, 6]) - expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ sessionID: parent.id }) + expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined() }), ) @@ -265,13 +263,13 @@ describe("SessionV2.create", () => { text: "First", resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id) + yield* SessionPending.promoteSteers(db, events, parent.id) const second = yield* session.prompt({ sessionID: parent.id, text: "Second", resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id) + yield* SessionPending.promoteSteers(db, events, parent.id) const assistantMessageID = SessionMessage.ID.create() const model = ModelV2.Ref.make({ id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }) yield* events.publish(SessionEvent.Step.Started, { @@ -416,7 +414,7 @@ describe("SessionV2.create", () => { text: "Hello", resume: false, }) - yield* SessionInput.promoteSteers(db, events, created.id) + yield* SessionPending.promoteSteers(db, events, created.id) expect( Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)), @@ -442,7 +440,7 @@ describe("SessionV2.create", () => { text: "Replay lifecycle", resume: false, }) - yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id) + yield* SessionPending.promoteSteers(sourceDb, sourceEvents, created.id) const serialized = (yield* sourceDb .select() .from(EventTable) @@ -480,7 +478,7 @@ describe("SessionV2.create", () => { expect(yield* store.get(created.id)).toBeUndefined() expect(yield* events.replayAll(serialized.slice(0, 2))).toBe(created.id) - expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ + expect(yield* SessionPending.find(db, admitted.id)).toMatchObject({ id: admitted.id, sessionID: created.id, type: "user", @@ -491,15 +489,7 @@ describe("SessionV2.create", () => { expect(yield* store.context(created.id)).toEqual([]) expect(yield* events.replayAll(serialized.slice(2))).toBe(created.id) - expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ - id: admitted.id, - sessionID: created.id, - type: "user", - data: { text: "Replay lifecycle" }, - delivery: "steer", - admittedSeq: 1, - promotedSeq: 2, - }) + expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined() expect(yield* store.context(created.id)).toMatchObject([ { id: admitted.id, type: "user", text: "Replay lifecycle" }, ]) diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index 7ddfa9b94b..e03fd3faff 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -20,11 +20,11 @@ import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" import { fromRow } from "@opencode-ai/core/session/info" -import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionPending } from "@opencode-ai/core/session/pending" import { Shell } from "@opencode-ai/schema/shell" import { InstructionCheckpointTable, - SessionInputTable, + SessionPendingTable, SessionMessageTable, SessionTable, } from "@opencode-ai/core/session/sql" @@ -77,7 +77,7 @@ describe("SessionProjector", () => { .run() const events = yield* EventV2.Service const inputID = SessionMessage.ID.make("msg_manual_compaction") - yield* SessionInput.admitCompaction(db, events, { id: inputID, sessionID }) + yield* SessionPending.admitCompaction(db, events, { id: inputID, sessionID }) yield* events.publish(SessionEvent.Compaction.Failed, { sessionID, @@ -85,7 +85,7 @@ describe("SessionProjector", () => { error: { type: "compaction.failed", message: "Auto compaction failed" }, }) - expect(yield* SessionInput.pendingCompaction(db, sessionID)).toMatchObject({ id: inputID }) + expect(yield* SessionPending.compaction(db, sessionID)).toMatchObject({ id: inputID }) }), ) @@ -316,7 +316,7 @@ describe("SessionProjector", () => { }).pipe(Effect.provide(sessionsLayer)), ) - it.effect("marks an inbox row promoted with the PromptPromoted event sequence", () => + it.effect("consumes the pending row and projects the message at promotion", () => Effect.gen(function* () { const { db } = yield* Database.Service yield* db @@ -338,7 +338,7 @@ describe("SessionProjector", () => { .pipe(Effect.orDie) const events = yield* EventV2.Service const id = SessionMessage.ID.make("msg_admitted") - const admitted = yield* SessionInput.admit(db, events, { + const admitted = yield* SessionPending.admit(db, events, { id, sessionID, input: { type: "user", data: { text: "promote me" }, delivery: "steer" }, @@ -351,8 +351,11 @@ describe("SessionProjector", () => { }) expect( - yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie), - ).toMatchObject({ promoted_seq: event.durable?.seq }) + yield* db.select().from(SessionPendingTable).where(eq(SessionPendingTable.id, id)).get().pipe(Effect.orDie), + ).toBeUndefined() + expect( + yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.id, id)).get().pipe(Effect.orDie), + ).toMatchObject({ session_id: sessionID, type: "user", seq: event.durable?.seq }) }), ) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index f9782170f4..7ea2633f62 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -21,8 +21,8 @@ import { SessionV2 } from "@opencode-ai/core/session" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" -import { SessionInput } from "@opencode-ai/core/session/input" -import { SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionPending } from "@opencode-ai/core/session/pending" +import { SessionPendingTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { testEffect } from "./lib/effect" @@ -81,11 +81,11 @@ const setup = Effect.gen(function* () { .pipe(Effect.orDie) }) -const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionInput.find(db, id)) +const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionPending.find(db, id)) const admittedCount = Database.Service.use(({ db }) => db .select() - .from(SessionInputTable) + .from(SessionPendingTable) .all() .pipe( Effect.orDie, @@ -201,7 +201,7 @@ describe("SessionV2.prompt", () => { text: "boundary", resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID) + yield* SessionPending.promoteSteers(db, events, sessionID) const stale = SessionMessage.ID.make("msg_stale_assistant") yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie) yield* events.publish(SessionEvent.RevertEvent.Staged, { @@ -218,7 +218,7 @@ describe("SessionV2.prompt", () => { (row) => row.id, ), ).not.toContainAnyValues([boundary.id, stale]) - expect(yield* SessionInput.find(db, boundary.id)).toBeUndefined() + expect(yield* SessionPending.find(db, boundary.id)).toBeUndefined() }), ) @@ -233,7 +233,7 @@ describe("SessionV2.prompt", () => { text: "boundary", resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID) + yield* SessionPending.promoteSteers(db, events, sessionID) yield* events.publish(SessionEvent.RevertEvent.Staged, { sessionID, revert: { messageID: boundary.id, files: [] }, @@ -243,11 +243,11 @@ describe("SessionV2.prompt", () => { const completion = yield* session.synthetic({ sessionID, text: "stale completion" }) expect(wakeCalls).toEqual([]) - expect(yield* SessionInput.find(db, completion.id)).toMatchObject({ type: "synthetic" }) + expect(yield* SessionPending.find(db, completion.id)).toMatchObject({ type: "synthetic" }) yield* session.revert.commit(sessionID) - expect(yield* SessionInput.find(db, completion.id)).toBeUndefined() + expect(yield* SessionPending.find(db, completion.id)).toBeUndefined() }), ) @@ -433,7 +433,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ sessionID, text: "First", resume: false }) yield* session.prompt({ sessionID, text: "Second", resume: false }) - yield* SessionInput.promoteSteers(db, events, sessionID) + yield* SessionPending.promoteSteers(db, events, sessionID) const streamed = Array.from(yield* Fiber.join(fiber)) expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([ @@ -610,12 +610,12 @@ describe("SessionV2.prompt", () => { }) yield* Effect.all( - [SessionInput.promoteSteers(db, events, sessionID), SessionInput.promoteSteers(db, events, sessionID)], + [SessionPending.promoteSteers(db, events, sessionID), SessionPending.promoteSteers(db, events, sessionID)], { concurrency: "unbounded" }, ) expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputPromoted.type, 1))).toBe(1) - expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 }) + expect(yield* admitted(messageID)).toBeUndefined() expect(yield* session.messages({ sessionID })).toMatchObject([ { id: messageID, type: "user", text: "Promote once" }, ]) @@ -645,7 +645,11 @@ describe("SessionV2.prompt", () => { .pipe(Effect.orDie) yield* events.remove(sessionID) - yield* db.delete(SessionInputTable).where(eq(SessionInputTable.session_id, sessionID)).run().pipe(Effect.orDie) + yield* db + .delete(SessionPendingTable) + .where(eq(SessionPendingTable.session_id, sessionID)) + .run() + .pipe(Effect.orDie) yield* db .delete(SessionMessageTable) .where(eq(SessionMessageTable.session_id, sessionID)) @@ -836,7 +840,7 @@ describe("SessionV2.prompt", () => { }, }) - yield* SessionInput.promoteSteers(db, events, sessionID) + yield* SessionPending.promoteSteers(db, events, sessionID) expect(yield* session.messages({ sessionID })).toMatchObject([ { @@ -861,14 +865,14 @@ describe("SessionV2.prompt", () => { const entries = yield* Effect.all([session.synthetic(input), session.synthetic(input)], { concurrency: "unbounded", }) - yield* SessionInput.promoteSteers(database.db, events, sessionID) + yield* SessionPending.promoteSteers(database.db, events, sessionID) const promotedRetry = yield* session.synthetic(input) const failure = yield* session.synthetic({ ...input, text: "Different completion" }).pipe(Effect.flip) expect(entries[1]).toEqual(entries[0]) - expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", promotedSeq: expect.any(Number) }) + expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", data: { text: "Completed" } }) expect(failure).toMatchObject({ _tag: "Session.SyntheticConflictError", sessionID, inputID: messageID }) - expect(yield* admittedCount).toBe(1) + expect(yield* admittedCount).toBe(0) expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputAdmitted.type, 1))).toBe(1) }), ) @@ -888,9 +892,9 @@ describe("SessionV2.prompt", () => { }) expect(input.delivery).toBe("queue") - expect(yield* SessionInput.promoteSteers(db, events, sessionID)).toBe(0) + expect(yield* SessionPending.promoteSteers(db, events, sessionID)).toBe(0) expect(yield* session.messages({ sessionID })).toEqual([]) - expect(yield* SessionInput.promoteNextQueued(db, events, sessionID)).toBe(true) + expect(yield* SessionPending.promoteNextQueued(db, events, sessionID)).toBe(true) expect(yield* session.messages({ sessionID })).toMatchObject([ { id: input.id, type: "synthetic", text: "Queued completion" }, ]) @@ -916,7 +920,7 @@ describe("SessionV2.prompt", () => { resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID) + yield* SessionPending.promoteSteers(db, events, sessionID) expect( (yield* session.messages({ sessionID, order: "asc" })).map((message) => @@ -926,3 +930,58 @@ describe("SessionV2.prompt", () => { }), ) }) + +describe("SessionV2.pending", () => { + it.effect("fails for an unknown session", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + expect(yield* session.pending(SessionV2.ID.make("ses_missing")).pipe(Effect.flip)).toMatchObject({ + _tag: "Session.NotFoundError", + }) + }), + ) + + it.effect("lists admitted work in admission order until promotion", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const { db } = yield* Database.Service + + const first = yield* session.prompt({ sessionID, text: "First steer", resume: false }) + const queued = yield* session.synthetic({ + sessionID, + text: "Queued completion", + delivery: "queue", + resume: false, + }) + const second = yield* session.prompt({ sessionID, text: "Second steer", resume: false }) + + expect(yield* session.pending(sessionID)).toMatchObject([ + { id: first.id, type: "user", delivery: "steer" }, + { id: queued.id, type: "synthetic", delivery: "queue" }, + { id: second.id, type: "user", delivery: "steer" }, + ]) + + yield* SessionPending.promoteSteers(db, events, sessionID) + expect(yield* session.pending(sessionID)).toMatchObject([{ id: queued.id, type: "synthetic" }]) + + yield* SessionPending.promoteNextQueued(db, events, sessionID) + expect(yield* session.pending(sessionID)).toEqual([]) + }), + ) + + it.effect("lists an unhandled compaction barrier until it settles", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const { db } = yield* Database.Service + + const barrier = yield* session.compact({ sessionID }) + expect(yield* session.pending(sessionID)).toMatchObject([{ id: barrier.id, type: "compaction" }]) + + yield* SessionPending.settleCompaction(db, { sessionID }) + expect(yield* session.pending(sessionID)).toEqual([]) + }), + ) +}) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 4465b9e350..86e9af37a9 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -27,7 +27,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionV2 } from "@opencode-ai/core/session" import { Snapshot } from "@opencode-ai/core/snapshot" import { SessionEvent } from "@opencode-ai/core/session/event" -import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionPending } from "@opencode-ai/core/session/pending" import { SessionMessage } from "@opencode-ai/core/session/message" import { Money } from "@opencode-ai/schema/money" import { SessionProjector } from "@opencode-ai/core/session/projector" @@ -47,7 +47,7 @@ import { ConfigCompaction } from "@opencode-ai/core/config/compaction" import { Tool } from "@opencode-ai/core/tool/tool" import { InstructionCheckpointTable, - SessionInputTable, + SessionPendingTable, SessionMessageTable, SessionTable, } from "@opencode-ai/core/session/sql" @@ -578,7 +578,7 @@ const replaySessionProjection = (id: SessionV2.ID) => .pipe(Effect.orDie) yield* events.remove(id) - yield* db.delete(SessionInputTable).where(eq(SessionInputTable.session_id, id)).run().pipe(Effect.orDie) + yield* db.delete(SessionPendingTable).where(eq(SessionPendingTable.session_id, id)).run().pipe(Effect.orDie) yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, id)).run().pipe(Effect.orDie) yield* events.replayAll( recorded.map((event) => ({ @@ -881,7 +881,7 @@ describe("SessionRunnerLLM", () => { expect(Exit.isFailure(exit)).toBe(true) if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(Instructions.InitializationBlocked) expect(requests).toHaveLength(0) - expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true) + expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true) expect( yield* db .select() @@ -924,7 +924,7 @@ describe("SessionRunnerLLM", () => { expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) expect(requests).toHaveLength(1) - expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true) + expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true) }), ) @@ -1462,7 +1462,7 @@ describe("SessionRunnerLLM", () => { const first = yield* session.compact({ sessionID }) const second = yield* session.compact({ sessionID }) expect(second.id).toBe(first.id) - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toMatchObject({ + expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toMatchObject({ id: first.id, }) expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toBeUndefined() @@ -1475,7 +1475,7 @@ describe("SessionRunnerLLM", () => { delivery: "queue", resume: false, }) - expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false) + expect(yield* SessionPending.has((yield* Database.Service).db, sessionID, "steer")).toBe(false) yield* Deferred.succeed(streamGate, undefined) yield* Fiber.join(active) @@ -1485,7 +1485,7 @@ describe("SessionRunnerLLM", () => { expect(userTexts(requests[2])).toContain("Steer after compaction") expect(userTexts(requests[2])).toContain("Completion after compaction") expect(userTexts(requests[3])).toContain("Queue after compaction") - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined() expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({ type: "compaction", status: "completed", @@ -1521,7 +1521,7 @@ describe("SessionRunnerLLM", () => { expect(requests).toHaveLength(3) expect(userTexts(requests[2])).toContain("Continue after failure") - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined() expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ type: "compaction", status: "failed", @@ -1542,7 +1542,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined() expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ type: "compaction", status: "failed", @@ -1566,7 +1566,7 @@ describe("SessionRunnerLLM", () => { expect(yield* Effect.exit(session.resume(sessionID))).toMatchObject({ _tag: "Failure" }) - expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined() expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ type: "compaction", status: "failed", @@ -2317,7 +2317,7 @@ describe("SessionRunnerLLM", () => { yield* session.interrupt(sessionID) expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) expect(requests).toHaveLength(1) - expect(yield* SessionInput.hasPending(db, sessionID, "queue")).toBe(true) + expect(yield* SessionPending.has(db, sessionID, "queue")).toBe(true) const resumed = yield* session.resume(sessionID).pipe(Effect.forkChild) while (requests.length < 2) yield* Effect.yieldNow yield* Deferred.succeed(streamGate, undefined) @@ -2350,7 +2350,7 @@ describe("SessionRunnerLLM", () => { yield* session.interrupt(sessionID) expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) expect(requests).toHaveLength(1) - expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true) + expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true) const resumed = yield* session.resume(sessionID).pipe(Effect.forkChild) while (requests.length < 2) yield* Effect.yieldNow @@ -2516,7 +2516,7 @@ describe("SessionRunnerLLM", () => { const session = yield* setup const events = yield* EventV2.Service yield* admit(session, "Recover interrupted tool") - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) + yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2573,7 +2573,7 @@ describe("SessionRunnerLLM", () => { const session = yield* setup const events = yield* EventV2.Service yield* admit(session, "Recover interrupted hosted tool") - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) + yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2624,7 +2624,7 @@ describe("SessionRunnerLLM", () => { const session = yield* setup const events = yield* EventV2.Service yield* admit(session, "Recover interrupted tool input") - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) + yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -3056,13 +3056,11 @@ describe("SessionRunnerLLM", () => { const releaseLateEvent = yield* Deferred.make() yield* registry.register({ permissionfail: permissionFail }) const events = yield* EventV2.Service - const permissionFailed = yield* events - .subscribe(SessionEvent.Tool.Failed) - .pipe( - Stream.filter((event) => event.data.sessionID === sessionID && event.data.callID === "call-permission"), - Stream.runHead, - Effect.forkScoped({ startImmediately: true }), - ) + const permissionFailed = yield* events.subscribe(SessionEvent.Tool.Failed).pipe( + Stream.filter((event) => event.data.sessionID === sessionID && event.data.callID === "call-permission"), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ) yield* admit(session, "Reject permission while another tool input streams") responseStream = Stream.concat( Stream.fromIterable([ diff --git a/packages/core/test/shared-schema.test.ts b/packages/core/test/shared-schema.test.ts index 8cbe9f9e2f..b4ee1ed494 100644 --- a/packages/core/test/shared-schema.test.ts +++ b/packages/core/test/shared-schema.test.ts @@ -12,7 +12,7 @@ import { ProjectDirectories } from "@opencode-ai/schema/project-directories" import { PermissionV1 } from "@opencode-ai/schema/permission-v1" import { Prompt } from "@opencode-ai/schema/prompt" import { Session } from "@opencode-ai/schema/session" -import { SessionInput } from "@opencode-ai/schema/session-input" +import { SessionPending } from "@opencode-ai/schema/session-pending" import { SessionMessage } from "@opencode-ai/schema/session-message" import { Workspace } from "@opencode-ai/schema/workspace" import { Command } from "@opencode-ai/schema/command" @@ -43,7 +43,7 @@ test("Core reuses the canonical shared schemas", async () => { corePty, coreProject, coreReference, - coreSessionInput, + coreSessionPending, coreSessionMessage, coreSkill, coreV2Schema, @@ -63,7 +63,7 @@ test("Core reuses the canonical shared schemas", async () => { import("@opencode-ai/core/pty"), import("@opencode-ai/core/project/schema"), import("@opencode-ai/core/reference"), - import("@opencode-ai/core/session/input"), + import("@opencode-ai/core/session/pending"), import("@opencode-ai/core/session/message"), import("@opencode-ai/core/skill"), import("@opencode-ai/core/v2-schema"), @@ -133,10 +133,10 @@ test("Core reuses the canonical shared schemas", async () => { [SessionV2.ID, Session.ID], [SessionV2.Info, Session.Info], [SessionV2.ListAnchor, Session.ListAnchor], - [coreSessionInput.Delivery, SessionInput.Delivery], - [coreSessionInput.Message, SessionInput.Message], - [coreSessionInput.User, SessionInput.User], - [coreSessionInput.Synthetic, SessionInput.Synthetic], + [coreSessionPending.Delivery, SessionPending.Delivery], + [coreSessionPending.Message, SessionPending.Message], + [coreSessionPending.User, SessionPending.User], + [coreSessionPending.Synthetic, SessionPending.Synthetic], [coreSessionMessage.ID, SessionMessage.ID], [coreSessionMessage.AssistantRetry, SessionMessage.AssistantRetry], [coreSessionMessage.AgentSelected, SessionMessage.AgentSelected], diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index e8d770d00a..661ef5d411 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -16,7 +16,7 @@ import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import { SessionV2 } from "@opencode-ai/core/session" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionExecution } from "@opencode-ai/core/session/execution" -import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionPending } from "@opencode-ai/core/session/pending" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" import { SessionStore } from "@opencode-ai/core/session/store" @@ -288,7 +288,7 @@ describe("SubagentTool", () => { const admission = Array.from(yield* Fiber.join(admitted))[0] expect(admission?.data.input.data.text).toContain(` message.type === "synthetic") expect(synthetic).toHaveLength(1) expect(synthetic[0]?.text).toContain(` ctx.session({ title: "Pending list owner" })) + .at((ctx) => ({ + path: route("/api/session/{sessionID}/pending", { sessionID: ctx.state.id }), + headers: ctx.headers(), + })) + .json(200, data(array)), http.protected .post("/api/session/{sessionID}/revert/stage", "v2.session.revert.stage") .at((ctx) => ({ diff --git a/packages/opencode/test/server/httpapi-session.test.ts b/packages/opencode/test/server/httpapi-session.test.ts index 4b3e7e3f92..fab8d4e57b 100644 --- a/packages/opencode/test/server/httpapi-session.test.ts +++ b/packages/opencode/test/server/httpapi-session.test.ts @@ -26,7 +26,7 @@ import { SessionPaths } from "../../src/server/routes/instance/httpapi/groups/se import { Session } from "@/session/session" import { MessageID, PartID, SessionID, type SessionID as SessionIDType } from "../../src/session/schema" import { Database } from "@opencode-ai/core/database/database" -import { SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionPendingTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionMessage } from "@opencode-ai/core/session/message" import { Agent } from "@opencode-ai/schema/agent" import { ModelV2 } from "@opencode-ai/core/model" @@ -566,18 +566,18 @@ describe("session HttpApi", () => { request(`/api/session/${session.id}/prompt`, { method: "POST", headers: { ...headers, "content-type": "application/json" }, - body: JSON.stringify({ id: "msg_http_prompt", prompt: { text: "hello" }, resume: false }), + body: JSON.stringify({ id: "msg_http_prompt", text: "hello", resume: false }), }) const first = yield* recordPrompt() const retried = yield* recordPrompt() - type PromptBody = { id: string; prompt: { text: string }; delivery: string; promotedSeq?: number } + type PromptBody = { id: string; data: { text: string }; delivery: string } const firstBody = yield* json<{ data: PromptBody }>(first) const retriedBody = yield* json<{ data: PromptBody }>(retried) expect(first.status).toBe(200) expect(retried.status).toBe(200) expect(retriedBody).toEqual(firstBody) expect(firstBody).toMatchObject({ - data: { id: "msg_http_prompt", prompt: { text: "hello" }, delivery: "steer" }, + data: { id: "msg_http_prompt", data: { text: "hello" }, delivery: "steer" }, }) const messages = yield* requestJson<{ data: PromptBody[] }>(`/api/session/${session.id}/message`, { @@ -587,8 +587,8 @@ describe("session HttpApi", () => { const admitted = yield* Database.Service.use(({ db }) => db .select() - .from(SessionInputTable) - .where(eq(SessionInputTable.id, SessionMessage.ID.make("msg_http_prompt"))) + .from(SessionPendingTable) + .where(eq(SessionPendingTable.id, SessionMessage.ID.make("msg_http_prompt"))) .get() .pipe(Effect.orDie), ) @@ -596,12 +596,11 @@ describe("session HttpApi", () => { id: "msg_http_prompt", session_id: session.id, delivery: "steer", - promoted_seq: null, }) const conflict = yield* request(`/api/session/${session.id}/prompt`, { method: "POST", headers: { ...headers, "content-type": "application/json" }, - body: JSON.stringify({ id: "msg_http_prompt", prompt: { text: "goodbye" } }), + body: JSON.stringify({ id: "msg_http_prompt", text: "goodbye" }), }) expect(conflict.status).toBe(409) expect(yield* responseJson(conflict)).toEqual({ @@ -614,7 +613,7 @@ describe("session HttpApi", () => { const wake = yield* request(`/api/session/${session.id}/prompt`, { method: "POST", headers: { ...headers, "content-type": "application/json" }, - body: JSON.stringify({ id: wakeID, prompt: { text: "hello again" } }), + body: JSON.stringify({ id: wakeID, text: "hello again" }), }) expect(wake.status).toBe(200) const message = yield* pollWithTimeout( diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 255ee420ab..19499b7709 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -1,5 +1,5 @@ import { SessionMessage } from "@opencode-ai/schema/session-message" -import { SessionInput } from "@opencode-ai/schema/session-input" +import { SessionPending } from "@opencode-ai/schema/session-pending" import { PromptInput } from "@opencode-ai/schema/prompt-input" import { Session } from "@opencode-ai/schema/session" import { InstructionEntry } from "@opencode-ai/schema/instruction-entry" @@ -294,11 +294,11 @@ export const makeSessionGroup = (sessionLo payload: Schema.Struct({ id: SessionMessage.ID.pipe(Schema.optional), ...PromptInput.Prompt.fields, - metadata: SessionInput.UserData.fields.metadata, - delivery: SessionInput.Delivery.pipe(Schema.optional), + metadata: SessionPending.UserData.fields.metadata, + delivery: SessionPending.Delivery.pipe(Schema.optional), resume: Schema.Boolean.pipe(Schema.optional), }), - success: Schema.Struct({ data: SessionInput.User }), + success: Schema.Struct({ data: SessionPending.User }), error: [ConflictError, InvalidRequestError, SessionNotFoundError], }) .middleware(sessionLocationMiddleware) @@ -321,10 +321,10 @@ export const makeSessionGroup = (sessionLo model: Model.Ref.pipe(Schema.optional), files: PromptInput.Prompt.fields.files, agents: PromptInput.Prompt.fields.agents, - delivery: SessionInput.Delivery.pipe(Schema.optional), + delivery: SessionPending.Delivery.pipe(Schema.optional), resume: Schema.Boolean.pipe(Schema.optional), }), - success: Schema.Struct({ data: SessionInput.User }), + success: Schema.Struct({ data: SessionPending.User }), error: [ConflictError, InvalidRequestError, SessionNotFoundError, CommandNotFoundError, CommandEvaluationError], }) .middleware(sessionLocationMiddleware) @@ -365,10 +365,10 @@ export const makeSessionGroup = (sessionLo text: Schema.String, description: Schema.String.pipe(Schema.optional), metadata: SessionMessage.Synthetic.fields.metadata, - delivery: SessionInput.Delivery.pipe(Schema.optional), + delivery: SessionPending.Delivery.pipe(Schema.optional), resume: Schema.Boolean.pipe(Schema.optional), }), - success: Schema.Struct({ data: SessionInput.Synthetic }), + success: Schema.Struct({ data: SessionPending.Synthetic }), error: [ConflictError, SessionNotFoundError], }) .middleware(sessionLocationMiddleware) @@ -404,7 +404,7 @@ export const makeSessionGroup = (sessionLo HttpApiEndpoint.post("session.compact", "/api/session/:sessionID/compact", { params: { sessionID: Session.ID }, payload: Schema.Struct({ id: SessionMessage.ID.pipe(Schema.optional) }), - success: Schema.Struct({ data: SessionInput.Compaction }), + success: Schema.Struct({ data: SessionPending.Compaction }), error: [ConflictError, SessionNotFoundError], }) .middleware(sessionLocationMiddleware) @@ -482,6 +482,22 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.get("session.pending.list", "/api/session/:sessionID/pending", { + params: { sessionID: Session.ID }, + success: Schema.Struct({ data: Schema.Array(SessionPending.Info) }), + error: SessionNotFoundError, + }) + .middleware(sessionLocationMiddleware) + .annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.pending.list", + summary: "List pending session work", + description: + "List durable admitted session work not yet visible in projected history, ordered by admission. Includes unpromoted user and synthetic inputs and unhandled compaction barriers. The runner owns consumption; items disappear once promoted or handled.", + }), + ), + ) .add( HttpApiEndpoint.get("session.instructions.entry.list", "/api/session/:sessionID/instructions/entries", { params: { sessionID: Session.ID }, diff --git a/packages/schema/src/index.ts b/packages/schema/src/index.ts index 56bf6fa3c5..7b09e92426 100644 --- a/packages/schema/src/index.ts +++ b/packages/schema/src/index.ts @@ -20,7 +20,7 @@ export { Provider } from "./provider.js" export { Reference } from "./reference.js" export { Session } from "./session.js" export { Vcs } from "./vcs.js" -export { SessionInput } from "./session-input.js" +export { SessionPending } from "./session-pending.js" export { SessionError } from "./session-error.js" export { SessionMessage } from "./session-message.js" export { Snapshot } from "./snapshot.js" diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 7d6e17e315..b2b6ec08ef 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -19,7 +19,7 @@ import { Skill as SkillSchema } from "./skill.js" import { Money } from "./money.js" import { Snapshot } from "./snapshot.js" import { TokenUsage } from "./token-usage.js" -import { SessionInput } from "./session-input.js" +import { SessionPending } from "./session-pending.js" export { FileAttachment } @@ -130,7 +130,7 @@ export const InputAdmitted = Event.durable({ schema: { ...Base, inputID: SessionMessage.ID, - input: SessionInput.Message, + input: SessionPending.Message, }, }) export type InputAdmitted = typeof InputAdmitted.Type diff --git a/packages/schema/src/session-input.ts b/packages/schema/src/session-pending.ts similarity index 75% rename from packages/schema/src/session-input.ts rename to packages/schema/src/session-pending.ts index 0b76d661e5..215c13c299 100644 --- a/packages/schema/src/session-input.ts +++ b/packages/schema/src/session-pending.ts @@ -1,4 +1,4 @@ -export * as SessionInput from "./session-input.js" +export * as SessionPending from "./session-pending.js" import { Schema } from "effect" import { optional } from "./schema.js" @@ -15,32 +15,32 @@ export interface UserData extends Schema.Schema.Type {} export const UserData = Schema.Struct({ ...Prompt.fields, metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional), -}).annotate({ identifier: "SessionInput.UserData" }) +}).annotate({ identifier: "SessionPending.UserData" }) export interface SyntheticData extends Schema.Schema.Type {} export const SyntheticData = Schema.Struct({ text: Schema.String, description: Schema.String.pipe(optional), metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional), -}).annotate({ identifier: "SessionInput.SyntheticData" }) +}).annotate({ identifier: "SessionPending.SyntheticData" }) export interface UserMessage extends Schema.Schema.Type {} export const UserMessage = Schema.Struct({ type: Schema.tag("user"), data: UserData, delivery: Delivery, -}).annotate({ identifier: "SessionInput.UserMessage" }) +}).annotate({ identifier: "SessionPending.UserMessage" }) export interface SyntheticMessage extends Schema.Schema.Type {} export const SyntheticMessage = Schema.Struct({ type: Schema.tag("synthetic"), data: SyntheticData, delivery: Delivery, -}).annotate({ identifier: "SessionInput.SyntheticMessage" }) +}).annotate({ identifier: "SessionPending.SyntheticMessage" }) export const Message = Schema.Union([UserMessage, SyntheticMessage]).pipe( Schema.toTaggedUnion("type"), - Schema.annotate({ identifier: "SessionInput.Message" }), + Schema.annotate({ identifier: "SessionPending.Message" }), ) export type Message = typeof Message.Type @@ -50,32 +50,27 @@ const Admitted = { sessionID: SessionID, timeCreated: DateTimeUtcFromMillis, } -const MessageLifecycle = { - ...Admitted, - promotedSeq: NonNegativeInt.pipe(optional), -} export interface User extends Schema.Schema.Type {} export const User = Schema.Struct({ - ...MessageLifecycle, + ...Admitted, ...UserMessage.fields, -}).annotate({ identifier: "SessionInput.User" }) +}).annotate({ identifier: "SessionPending.User" }) export interface Synthetic extends Schema.Schema.Type {} export const Synthetic = Schema.Struct({ - ...MessageLifecycle, + ...Admitted, ...SyntheticMessage.fields, -}).annotate({ identifier: "SessionInput.Synthetic" }) +}).annotate({ identifier: "SessionPending.Synthetic" }) export interface Compaction extends Schema.Schema.Type {} export const Compaction = Schema.Struct({ ...Admitted, type: Schema.tag("compaction"), - handledSeq: NonNegativeInt.pipe(optional), -}).annotate({ identifier: "SessionInput.Compaction" }) +}).annotate({ identifier: "SessionPending.Compaction" }) export const Info = Schema.Union([User, Synthetic, Compaction]).pipe( Schema.toTaggedUnion("type"), - Schema.annotate({ identifier: "SessionInput.Info" }), + Schema.annotate({ identifier: "SessionPending.Info" }), ) export type Info = typeof Info.Type diff --git a/packages/schema/test/contract-hygiene.test.ts b/packages/schema/test/contract-hygiene.test.ts index 140be20cba..e34b3b6f6e 100644 --- a/packages/schema/test/contract-hygiene.test.ts +++ b/packages/schema/test/contract-hygiene.test.ts @@ -10,7 +10,7 @@ import { Pty } from "../src/pty.js" import { Question } from "../src/question.js" import { Session } from "../src/session.js" import { SessionMessage } from "../src/session-message.js" -import { SessionInput } from "../src/session-input.js" +import { SessionPending } from "../src/session-pending.js" import { FileDiff } from "../src/file-diff.js" import { Money } from "../src/money.js" import { Skill } from "../src/skill.js" @@ -39,7 +39,7 @@ describe("contract hygiene", () => { expect(Schema.encodeSync(Value)({ value: 1 })).toEqual({ value: "1" }) expect(Schema.encodeSync(Value)({ value: undefined })).toEqual({}) expect( - Schema.encodeSync(SessionInput.SyntheticData)({ + Schema.encodeSync(SessionPending.SyntheticData)({ text: "completed", description: undefined, metadata: undefined, @@ -89,10 +89,10 @@ describe("contract hygiene", () => { Pty.Info, Session.ListAnchor, Session.Revert, - SessionInput.UserData, - SessionInput.SyntheticData, - SessionInput.User, - SessionInput.Synthetic, + SessionPending.UserData, + SessionPending.SyntheticData, + SessionPending.User, + SessionPending.Synthetic, ].map((schema) => schema.ast.annotations?.identifier) expect(identifiers.every((identifier) => typeof identifier === "string")).toBe(true) @@ -134,7 +134,7 @@ describe("contract hygiene", () => { test("reviewed session contracts use their canonical current shapes", () => { expect(SessionMessage.Info.ast.annotations?.identifier).toBe("Session.Message.Info") - expect(SessionInput.Info.ast.annotations?.identifier).toBe("SessionInput.Info") + expect(SessionPending.Info.ast.annotations?.identifier).toBe("SessionPending.Info") expect(Money.USD).not.toBe(Money.USDPerMillionTokens) expect( FileDiff.Info.make({ file: "src/index.ts", patch: "@@", additions: 1, deletions: 0, status: "modified" }), diff --git a/packages/sdk-next/package.json b/packages/sdk-next/package.json index 3b092e1745..8ef199b680 100644 --- a/packages/sdk-next/package.json +++ b/packages/sdk-next/package.json @@ -20,6 +20,8 @@ "effect": "catalog:" }, "devDependencies": { + "@opencode-ai/httpapi-codegen": "workspace:*", + "@opencode-ai/protocol": "workspace:*", "@tsconfig/bun": "catalog:", "@types/bun": "catalog:", "@typescript/native-preview": "catalog:" diff --git a/packages/sdk-next/src/index.ts b/packages/sdk-next/src/index.ts index d8ce5f656b..a9d91eeec3 100644 --- a/packages/sdk-next/src/index.ts +++ b/packages/sdk-next/src/index.ts @@ -22,6 +22,6 @@ export { Question } from "@opencode-ai/schema/question" export { Reference } from "@opencode-ai/schema/reference" export { AbsolutePath, RelativePath } from "@opencode-ai/schema/schema" export { Session } from "@opencode-ai/schema/session" -export { SessionInput } from "@opencode-ai/schema/session-input" +export { SessionPending } from "@opencode-ai/schema/session-pending" export { SessionMessage } from "@opencode-ai/schema/session-message" export { Skill } from "@opencode-ai/schema/skill" diff --git a/packages/sdk-next/test/contract-identity.test.ts b/packages/sdk-next/test/contract-identity.test.ts index 830fe5f43b..4aad96f13a 100644 --- a/packages/sdk-next/test/contract-identity.test.ts +++ b/packages/sdk-next/test/contract-identity.test.ts @@ -1,7 +1,23 @@ import { expect, test } from "bun:test" +import { AgentV2 } from "@opencode-ai/core/agent" +import { Location as CoreLocation } from "@opencode-ai/core/location" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProjectV2 } from "@opencode-ai/core/project" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionPending as CoreSessionPending } from "@opencode-ai/core/session/pending" +import { SessionMessage as CoreSessionMessage } from "@opencode-ai/core/session/message" import { Agent } from "@opencode-ai/schema/agent" +import { Location } from "@opencode-ai/schema/location" import { Model } from "@opencode-ai/schema/model" +import { Project } from "@opencode-ai/schema/project" +import { Provider } from "@opencode-ai/schema/provider" import { Session } from "@opencode-ai/schema/session" +import { SessionPending } from "@opencode-ai/schema/session-pending" +import { SessionMessage } from "@opencode-ai/schema/session-message" +import { Workspace } from "@opencode-ai/schema/workspace" +import { Api } from "@opencode-ai/server/api" +import { ClientApi, groupNames, promiseOmitEndpoints } from "@opencode-ai/protocol/client" +import { compile, emitPromise } from "@opencode-ai/httpapi-codegen" const SDK = await import("../src/index") @@ -32,9 +48,38 @@ test("re-exports canonical contracts directly from Schema", () => { "Reference", "RelativePath", "Session", - "SessionInput", "SessionMessage", + "SessionPending", "Skill", "Tool", ]) }) + +test("Core and Server reuse the authoritative Schema and Protocol values", () => { + expect(AgentV2.ID).toBe(Agent.ID) + expect(CoreLocation.Ref).toBe(Location.Ref) + expect(ModelV2.Ref).toBe(Model.Ref) + expect(SessionV2.Info).toBe(Session.Info) + expect(ProjectV2.Current).toBe(Project.Current) + expect(ProjectV2.Directory).toBe(Project.Directory) + expect(ProjectV2.Directories).toBe(Project.Directories) + expect(CoreSessionPending.Message).toBe(SessionPending.Message) + expect(CoreSessionPending.User).toBe(SessionPending.User) + expect(CoreSessionPending.Synthetic).toBe(SessionPending.Synthetic) + expect(CoreSessionMessage.Info).toBe(SessionMessage.Info) + expect(CoreSessionMessage.AssistantText).toBe(SessionMessage.AssistantText) + expect(Api.groups["server.session"].identifier).toBe("server.session") + expect(Api.groups["server.project"].identifier).toBe("server.project") + expect(Object.keys(ClientApi.groups)).toEqual(Object.keys(Api.groups)) + expect(Session.ID.create()).toStartWith("ses_") + expect(String(Project.ID.global)).toBe("global") + expect(String(Provider.ID.anthropic)).toBe("anthropic") + expect(Workspace.ID.create()).toStartWith("wrk_") +}) + +test("client and Server contracts generate identically", () => { + const server = compile(Api, { groupNames, omitEndpoints: promiseOmitEndpoints }) + const client = compile(ClientApi, { groupNames, omitEndpoints: promiseOmitEndpoints }) + + expect(emitPromise(client)).toEqual(emitPromise(server)) +}) diff --git a/packages/sdk-next/test/embedded.test.ts b/packages/sdk-next/test/embedded.test.ts index 482f8c9501..2214ebf873 100644 --- a/packages/sdk-next/test/embedded.test.ts +++ b/packages/sdk-next/test/embedded.test.ts @@ -229,6 +229,7 @@ it.live( resume: false, }) const context = yield* opencode.sessions.context({ sessionID: id }) + const pendingAfterAdmit = yield* opencode.sessions.pending.list({ sessionID: id }) yield* opencode.sessions.instructions.entry.put({ sessionID: id, key: "deploy-target", value: "production" }) yield* opencode.sessions.instructions.entry.put({ sessionID: id, key: "flags", value: { beta: true } }) const contextEntries = yield* opencode.sessions.instructions.entry.list({ sessionID: id }) @@ -245,6 +246,7 @@ it.live( Effect.map(Option.getOrThrow), ) const wakeContext = yield* opencode.sessions.context({ sessionID: id }) + const pendingAfterPromote = yield* opencode.sessions.pending.list({ sessionID: id }) const event = yield* opencode.sessions.log({ sessionID: id }).pipe( Stream.filter((item) => item.type !== "log.synced"), Stream.take(1), @@ -264,6 +266,7 @@ it.live( opencode.sessions.interrupt({ sessionID: missingSessionID }).pipe(Effect.flip), opencode.sessions.message({ sessionID: missingSessionID, messageID: modelMessage.id }).pipe(Effect.flip), opencode.sessions.instructions.entry.list({ sessionID: missingSessionID }).pipe(Effect.flip), + opencode.sessions.pending.list({ sessionID: missingSessionID }).pipe(Effect.flip), ], { concurrency: "unbounded" }, ) @@ -280,7 +283,11 @@ it.live( expect(page.data.some((session) => session.id === id)).toBe(true) expect(active).toEqual({}) expect(admitted.sessionID).toBe(id) + expect(pendingAfterAdmit).toContainEqual( + expect.objectContaining({ id: admitted.id, type: "user", delivery: "steer" }), + ) expect(prompted.type).toBe("session.input.promoted") + expect(pendingAfterPromote.map((item) => item.id)).not.toContainAnyValues([admitted.id, wake.id]) expect(wakeContext).toContainEqual(expect.objectContaining({ id: wake.id, type: "user" })) expect(contextEntries).toEqual([ { key: "deploy-target", value: "production" }, @@ -295,6 +302,7 @@ it.live( "SessionNotFoundError", "SessionNotFoundError", "SessionNotFoundError", + "SessionNotFoundError", ]) expect(missingMessage._tag).toBe("MessageNotFoundError") }), diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index c1ecda15ea..9d1c3a96e2 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -405,6 +405,8 @@ import type { V2SessionMessageResponses, V2SessionMoveErrors, V2SessionMoveResponses, + V2SessionPendingListErrors, + V2SessionPendingListResponses, V2SessionPermissionCreateErrors, V2SessionPermissionCreateResponses, V2SessionPermissionGetErrors, @@ -5251,6 +5253,31 @@ export class Revert extends HeyApiClient { } } +export class Pending extends HeyApiClient { + /** + * List pending session work + * + * List durable admitted session work not yet visible in projected history, ordered by admission. Includes unpromoted user and synthetic inputs and unhandled compaction barriers. The runner owns consumption; items disappear once promoted or handled. + */ + public list( + parameters: { + sessionID: string + }, + options?: Options, + ) { + const params = buildClientParams([parameters], [{ args: [{ in: "path", key: "sessionID" }] }]) + return (options?.client ?? this.client).get< + V2SessionPendingListResponses, + V2SessionPendingListErrors, + ThrowOnError + >({ + url: "/api/session/{sessionID}/pending", + ...options, + ...params, + }) + } +} + export class Entry extends HeyApiClient { /** * List instruction entries @@ -6524,6 +6551,11 @@ export class Session3 extends HeyApiClient { return (this._revert ??= new Revert({ client: this.client })) } + private _pending?: Pending + get pending(): Pending { + return (this._pending ??= new Pending({ client: this.client })) + } + private _instructions?: Instructions get instructions(): Instructions { return (this._instructions ??= new Instructions({ client: this.client })) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 92b9514db5..9baf54e1b7 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -846,7 +846,7 @@ export type GlobalEvent = { properties: { sessionID: string inputID: string - input: SessionInputMessage + input: SessionPendingMessage } } | { @@ -3352,7 +3352,7 @@ export type PromptAgentAttachment = { mention?: PromptMention } -export type SessionInputUserData = { +export type SessionPendingUserData = { text: string files?: Array agents?: Array @@ -3361,13 +3361,13 @@ export type SessionInputUserData = { } } -export type SessionInputUserMessage = { +export type SessionPendingUserMessage = { type: "user" - data: SessionInputUserData + data: SessionPendingUserData delivery: "steer" | "queue" } -export type SessionInputSyntheticData = { +export type SessionPendingSyntheticData = { text: string description?: string metadata?: { @@ -3375,13 +3375,13 @@ export type SessionInputSyntheticData = { } } -export type SessionInputSyntheticMessage = { +export type SessionPendingSyntheticMessage = { type: "synthetic" - data: SessionInputSyntheticData + data: SessionPendingSyntheticData delivery: "steer" | "queue" } -export type SessionInputMessage = SessionInputUserMessage | SessionInputSyntheticMessage +export type SessionPendingMessage = SessionPendingUserMessage | SessionPendingSyntheticMessage export type SessionStructuredError = { type: string @@ -3817,7 +3817,7 @@ export type SyncEventSessionInputAdmitted = { data: { sessionID: string inputID: string - input: SessionInputMessage + input: SessionPendingMessage } } } @@ -4450,35 +4450,32 @@ export type PromptInputFileAttachment = { mention?: PromptMention } -export type SessionInputUser = { +export type SessionPendingUser = { admittedSeq: number id: string sessionID: string timeCreated: number - promotedSeq?: number type: "user" - data: SessionInputUserData + data: SessionPendingUserData delivery: "steer" | "queue" } -export type SessionInputSynthetic = { +export type SessionPendingSynthetic = { admittedSeq: number id: string sessionID: string timeCreated: number - promotedSeq?: number type: "synthetic" - data: SessionInputSyntheticData + data: SessionPendingSyntheticData delivery: "steer" | "queue" } -export type SessionInputCompaction = { +export type SessionPendingCompaction = { admittedSeq: number id: string sessionID: string timeCreated: number type: "compaction" - handledSeq?: number } export type SessionMessageAgentSelected = { @@ -4747,6 +4744,8 @@ export type SessionMessageInfo = | SessionMessageAssistant | SessionMessageCompaction +export type SessionPendingInfo = SessionPendingUser | SessionPendingSynthetic | SessionPendingCompaction + export type InstructionEntryKey = string export type InstructionEntryInfo = { @@ -4904,7 +4903,7 @@ export type SessionInputAdmitted = { data: { sessionID: string inputID: string - input: SessionInputMessage + input: SessionPendingMessage } } @@ -7185,7 +7184,7 @@ export type EventSessionInputAdmitted = { properties: { sessionID: string inputID: string - input: SessionInputMessage + input: SessionPendingMessage } } @@ -8778,35 +8777,32 @@ export type SessionInfoV2 = { export type PromptBase64V2 = string -export type SessionInputUserV2 = { +export type SessionPendingUserV2 = { admittedSeq: number id: string sessionID: string timeCreated: number - promotedSeq?: number type: "user" - data: SessionInputUserData + data: SessionPendingUserData delivery: "steer" | "queue" } -export type SessionInputSyntheticV2 = { +export type SessionPendingSyntheticV2 = { admittedSeq: number id: string sessionID: string timeCreated: number - promotedSeq?: number type: "synthetic" - data: SessionInputSyntheticData + data: SessionPendingSyntheticData delivery: "steer" | "queue" } -export type SessionInputCompactionV2 = { +export type SessionPendingCompactionV2 = { admittedSeq: number id: string sessionID: string timeCreated: number type: "compaction" - handledSeq?: number } export type SessionMessageAgentSelectedV2 = { @@ -9123,7 +9119,7 @@ export type SessionInputPromotedV2 = { } } -export type SessionInputUserData1 = { +export type SessionPendingUserData1 = { text: string files?: Array agents?: Array @@ -9132,13 +9128,13 @@ export type SessionInputUserData1 = { } } -export type SessionInputUserMessageV2 = { +export type SessionPendingUserMessageV2 = { type: "user" - data: SessionInputUserData1 + data: SessionPendingUserData1 delivery: "steer" | "queue" } -export type SessionInputSyntheticData1 = { +export type SessionPendingSyntheticData1 = { text: string description?: string metadata?: { @@ -9146,9 +9142,9 @@ export type SessionInputSyntheticData1 = { } } -export type SessionInputSyntheticMessageV2 = { +export type SessionPendingSyntheticMessageV2 = { type: "synthetic" - data: SessionInputSyntheticData1 + data: SessionPendingSyntheticData1 delivery: "steer" | "queue" } @@ -9168,7 +9164,7 @@ export type SessionInputAdmittedV2 = { data: { sessionID: string inputID: string - input: SessionInputMessage + input: SessionPendingMessage } } @@ -15683,7 +15679,7 @@ export type V2SessionPromptResponses = { * Success */ 200: { - data: SessionInputUserV2 + data: SessionPendingUserV2 } } @@ -15738,7 +15734,7 @@ export type V2SessionCommandResponses = { * Success */ 200: { - data: SessionInputUserV2 + data: SessionPendingUserV2 } } @@ -15827,7 +15823,7 @@ export type V2SessionSyntheticResponses = { * Success */ 200: { - data: SessionInputSyntheticV2 + data: SessionPendingSyntheticV2 } } @@ -15908,7 +15904,7 @@ export type V2SessionCompactResponses = { * Success */ 200: { - data: SessionInputCompactionV2 + data: SessionPendingCompactionV2 } } @@ -16124,6 +16120,43 @@ export type V2SessionContextResponses = { export type V2SessionContextResponse = V2SessionContextResponses[keyof V2SessionContextResponses] +export type V2SessionPendingListData = { + body?: never + path: { + sessionID: string + } + query?: never + url: "/api/session/{sessionID}/pending" +} + +export type V2SessionPendingListErrors = { + /** + * InvalidRequestError + */ + 400: InvalidRequestErrorV2 + /** + * UnauthorizedError + */ + 401: UnauthorizedError + /** + * SessionNotFoundError + */ + 404: SessionNotFoundError +} + +export type V2SessionPendingListError = V2SessionPendingListErrors[keyof V2SessionPendingListErrors] + +export type V2SessionPendingListResponses = { + /** + * Success + */ + 200: { + data: Array + } +} + +export type V2SessionPendingListResponse = V2SessionPendingListResponses[keyof V2SessionPendingListResponses] + export type V2SessionInstructionsEntryListData = { body?: never path: { diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 8bb4694363..b2c2e7eb10 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -600,6 +600,23 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl } }), ) + .handle( + "session.pending.list", + Effect.fn(function* (ctx) { + return { + data: yield* session.pending(ctx.params.sessionID).pipe( + Effect.catchTag("Session.NotFoundError", (error) => + Effect.fail( + new SessionNotFoundError({ + sessionID: error.sessionID, + message: `Session not found: ${error.sessionID}`, + }), + ), + ), + ), + } + }), + ) .handle( "session.instructions.entry.list", Effect.fn(function* (ctx) { diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index 3821248161..3b5428572c 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -1,5 +1,18 @@ # V2 Schema Changelog +## 2026-07-09: Make Session Input Storage Pending-Only And Rename It To Session Pending + +- Rename the `SessionInput` schema namespace to `SessionPending` and the `session_input` table to `session_pending`. +- Make the table pending-only: promotion consumes the user or synthetic row in the same event transaction that projects the message, and compaction settlement deletes the barrier row. Drop the `promoted_seq` column and the retained `promotedSeq`/`handledSeq` wire fields. +- Add `GET /api/session/:sessionID/pending` (`v2.session.pending.list`) returning durable admitted work not yet visible in projected history, ordered by admission. +- Reconcile exact retry of an already-promoted input against the projected `session_message` row plus the durable `session.input.admitted` event instead of a retained row. A projected message without an admitted event in the aggregate (for example fork-copied history) is conflicting reuse. Reusing a settled compaction ID admits a fresh barrier instead of reconciling; the worst case is one redundant compaction on a retried request. + +Compatibility: + +- `20260709190621_session_pending_table` drops `session_input` (including consumed ledger rows, any in-flight pending work, and whatever historical index variant the database carried) and creates the empty `session_pending` table. V2 storage is beta; no compatibility or data retention is attempted. +- Durable event names and payloads are unchanged; `session.input.admitted` still records the full admitted message including delivery. +- Promise, Effect, and legacy JavaScript SDK surfaces are regenerated; `SessionInput*` generated schema names become `SessionPending*` while event-derived names keep their `session.input.*` vocabulary. + ## 2026-07-05: Rename Session Context Contracts To Instructions - Rename the System Context algebra to `Instructions`, API-managed `SessionContextEntry` records to `InstructionEntry`, and the session-owned context checkpoint to `InstructionCheckpoint`.