From 69d95f29d4777eb952a3adb138115d5561c43f94 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 4 Jul 2026 20:08:41 -0400 Subject: [PATCH 1/3] feat(core): add durable compaction barrier --- packages/client/src/effect/api/api.ts | 7 +- .../client/src/effect/generated/client.ts | 10 +- .../client/src/promise/generated/client.ts | 11 +- .../client/src/promise/generated/types.ts | 73 +++++- packages/client/test/effect.test.ts | 14 + packages/client/test/promise.test.ts | 12 + packages/core/schema.json | 36 ++- packages/core/src/database/migration.gen.ts | 1 + .../20260704161518_durable_session_inbox.ts | 43 +++ packages/core/src/database/schema.gen.ts | 10 +- packages/core/src/session.ts | 35 ++- packages/core/src/session/compaction.ts | 4 +- packages/core/src/session/history.ts | 10 +- packages/core/src/session/input.ts | 246 ++++++++++++++---- packages/core/src/session/message-updater.ts | 76 +++++- packages/core/src/session/projector.ts | 65 ++++- packages/core/src/session/runner/llm.ts | 44 +++- .../core/src/session/runner/to-llm-message.ts | 1 + packages/core/src/session/sql.ts | 12 +- packages/core/test/database-migration.test.ts | 39 ++- packages/core/test/session-compact.test.ts | 23 +- packages/core/test/session-prompt.test.ts | 4 +- .../core/test/session-runner-message.test.ts | 5 +- packages/core/test/session-runner.test.ts | 91 +++++++ packages/protocol/src/groups/session.ts | 7 +- packages/schema/src/session-event.ts | 19 ++ packages/schema/src/session-input.ts | 14 + packages/schema/src/session-message.ts | 1 + packages/schema/test/event-manifest.test.ts | 2 + packages/sdk/js/src/v2/gen/sdk.gen.ts | 20 +- packages/sdk/js/src/v2/gen/types.gen.ts | 189 +++++++++++++- packages/server/src/handlers/session.ts | 52 ++-- packages/tui/src/context/data.tsx | 53 ++++ packages/tui/src/routes/session/index.tsx | 67 ++++- packages/tui/src/routes/session/rows.ts | 33 ++- packages/tui/test/cli/tui/data.test.tsx | 40 +++ .../tui/test/cli/tui/session-rows.test.ts | 29 +++ specs/v2/session.md | 5 +- 38 files changed, 1209 insertions(+), 194 deletions(-) create mode 100644 packages/core/src/database/migration/20260704161518_durable_session_inbox.ts diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 3f4c852be198..af00c6f50766 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -163,8 +163,11 @@ export type Endpoint4_12Output = EffectValue = (input: Endpoint4_12Input) => Effect.Effect type Endpoint4_13Request = Parameters[0] -export type Endpoint4_13Input = { readonly sessionID: Endpoint4_13Request["params"]["sessionID"] } -export type Endpoint4_13Output = EffectValue> +export type Endpoint4_13Input = { + readonly sessionID: Endpoint4_13Request["params"]["sessionID"] + readonly id?: Endpoint4_13Request["payload"]["id"] +} +export type Endpoint4_13Output = EffectValue>["data"] export type SessionCompactOperation = (input: Endpoint4_13Input) => Effect.Effect type Endpoint4_14Request = Parameters[0] diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index a9a91eb8c617..85b7986ab565 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -221,9 +221,15 @@ const Endpoint4_12 = (raw: RawClient["server.session"]) => (input: Endpoint4_12I }).pipe(Effect.mapError(mapClientError)) type Endpoint4_13Request = Parameters[0] -type Endpoint4_13Input = { readonly sessionID: Endpoint4_13Request["params"]["sessionID"] } +type Endpoint4_13Input = { + readonly sessionID: Endpoint4_13Request["params"]["sessionID"] + readonly id?: Endpoint4_13Request["payload"]["id"] +} const Endpoint4_13 = (raw: RawClient["server.session"]) => (input: Endpoint4_13Input) => - raw["session.compact"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) + raw["session.compact"]({ params: { sessionID: input["sessionID"] }, payload: { id: input["id"] } }).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ) type Endpoint4_14Request = Parameters[0] type Endpoint4_14Input = { readonly sessionID: Endpoint4_14Request["params"]["sessionID"] } diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 79368764c749..f1ebe6266ad8 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -540,16 +540,17 @@ export function make(options: ClientOptions) { requestOptions, ), compact: (input: SessionCompactInput, requestOptions?: RequestOptions) => - request( + request<{ readonly data: SessionCompactOutput }>( { method: "POST", path: `/api/session/${encodeURIComponent(input.sessionID)}/compact`, - successStatus: 204, - declaredStatuses: [404, 409, 503, 500, 400, 401], - empty: true, + body: { id: input["id"] }, + successStatus: 200, + declaredStatuses: [409, 404, 400, 401], + empty: false, }, requestOptions, - ), + ).then((value) => value.data), wait: (input: SessionWaitInput, requestOptions?: RequestOptions) => request( { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 62320520f99f..7804428e3575 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -74,14 +74,6 @@ export type SkillNotFoundError = { export const isSkillNotFoundError = (value: unknown): value is SkillNotFoundError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "SkillNotFoundError" -export type SessionBusyError = { - readonly _tag: "SessionBusyError" - readonly sessionID: string - readonly message: string -} -export const isSessionBusyError = (value: unknown): value is SessionBusyError => - typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "SessionBusyError" - export type ServiceUnavailableError = { readonly _tag: "ServiceUnavailableError" readonly message: string @@ -90,6 +82,14 @@ export type ServiceUnavailableError = { export const isServiceUnavailableError = (value: unknown): value is ServiceUnavailableError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "ServiceUnavailableError" +export type SessionBusyError = { + readonly _tag: "SessionBusyError" + readonly sessionID: string + readonly message: string +} +export const isSessionBusyError = (value: unknown): value is SessionBusyError => + typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "SessionBusyError" + export type UnknownError = { readonly _tag: "UnknownError" readonly message: string @@ -600,6 +600,7 @@ export type SessionPromptInput = { export type SessionPromptOutput = { readonly data: { + readonly type: "prompt" readonly admittedSeq: number readonly id: string readonly sessionID: string @@ -800,6 +801,7 @@ export type SessionCommandInput = { export type SessionCommandOutput = { readonly data: { + readonly type: "prompt" readonly admittedSeq: number readonly id: string readonly sessionID: string @@ -873,9 +875,21 @@ export type SessionShellInput = { export type SessionShellOutput = void -export type SessionCompactInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } +export type SessionCompactInput = { + readonly sessionID: { readonly sessionID: string }["sessionID"] + readonly id?: { readonly id?: string | undefined }["id"] +} -export type SessionCompactOutput = void +export type SessionCompactOutput = { + readonly data: { + readonly type: "compaction" + readonly admittedSeq: number + readonly id: string + readonly sessionID: string + readonly timeCreated: number + readonly handledSeq?: number + } +}["data"] export type SessionWaitInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } @@ -1087,6 +1101,7 @@ export type SessionContextOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -1555,6 +1570,15 @@ export type SessionLogOutput = readonly error: { readonly type: string; readonly message: string } } } + | { + readonly id: string + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.compaction.admitted" + readonly durable: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly sessionID: string; readonly inputID: string } + } | { readonly id: string readonly created: number @@ -1578,6 +1602,15 @@ export type SessionLogOutput = readonly recent: string } } + | { + readonly id: string + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.compaction.failed" + readonly durable: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly sessionID: string } + } | { readonly id: string readonly created: number @@ -1810,6 +1843,7 @@ export type SessionMessageOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -2012,6 +2046,7 @@ export type MessageListOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -4894,6 +4929,15 @@ export type EventSubscribeOutput = readonly error: { readonly type: string; readonly message: string } } } + | { + readonly id: string + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.compaction.admitted" + readonly durable: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly sessionID: string; readonly inputID: string } + } | { readonly id: string readonly created: number @@ -4925,6 +4969,15 @@ export type EventSubscribeOutput = readonly recent: string } } + | { + readonly id: string + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.compaction.failed" + readonly durable: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly sessionID: string } + } | { readonly id: string readonly created: number diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index 2077c27ca5f5..40765d7be89e 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -89,6 +89,9 @@ test("session methods retain decoded Effect inputs and outputs", async () => { if (url.includes("/prompt")) { return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(admission))) } + if (url.endsWith("/compact")) { + return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(compactionAdmission))) + } if (url.includes("/context")) { return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ data: [] }))) } @@ -209,6 +212,7 @@ const session = { const admission = { data: { + type: "prompt", admittedSeq: 0, id: "msg_test", sessionID: "ses_test", @@ -218,6 +222,16 @@ const admission = { }, } +const compactionAdmission = { + data: { + type: "compaction", + admittedSeq: 1, + id: "msg_compaction", + sessionID: "ses_test", + timeCreated: 1_717_171_717_000, + }, +} + const modelSwitchedMessage = { id: "msg_model", type: "model-switched", diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index d445a69c0833..ca2d5bba1b36 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -193,6 +193,7 @@ test("session methods use the public HTTP contract", async () => { }) } if (url.includes("/prompt")) return Response.json(admission) + if (url.endsWith("/compact")) return Response.json(compactionAdmission) if (url.includes("/context")) return Response.json({ data: [] }) if (url.includes("/message/")) return Response.json({ data: modelSwitchedMessage }) if (url.endsWith("/api/session/active")) @@ -308,6 +309,7 @@ const session = { const admission = { data: { + type: "prompt", admittedSeq: 0, id: "msg_test", sessionID: "ses_test", @@ -317,6 +319,16 @@ const admission = { }, } +const compactionAdmission = { + data: { + type: "compaction", + admittedSeq: 1, + id: "msg_compaction", + sessionID: "ses_test", + timeCreated: 1_717_171_717_000, + }, +} + const modelSwitchedMessage = { id: "msg_model", type: "model-switched", diff --git a/packages/core/schema.json b/packages/core/schema.json index d9e0b20b22f7..c0f519cfadde 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,9 +1,9 @@ { "version": "7", "dialect": "sqlite", - "id": "96e9fe64-d810-4102-8f79-3317a88bb6d2", + "id": "afa00750-fad8-473a-b877-13ff35ea3411", "prevIds": [ - "22e57fed-b9b8-4e94-a3b4-f94bece680a8" + "96e9fe64-d810-4102-8f79-3317a88bb6d2" ], "ddl": [ { @@ -1012,13 +1012,23 @@ "autoincrement": false, "default": null, "generated": null, + "name": "type", + "entityType": "columns", + "table": "session_input" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, "name": "prompt", "entityType": "columns", "table": "session_input" }, { "type": "text", - "notNull": true, + "notNull": false, "autoincrement": false, "default": null, "generated": null, @@ -2066,6 +2076,10 @@ "value": "promoted_seq", "isExpression": false }, + { + "value": "type", + "isExpression": false + }, { "value": "delivery", "isExpression": false @@ -2078,7 +2092,21 @@ "isUnique": false, "where": null, "origin": "manual", - "name": "session_input_session_pending_delivery_seq_idx", + "name": "session_input_session_pending_type_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" }, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 80cb22fb94e2..e82a2a6e1e0e 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -45,5 +45,6 @@ export const migrations = ( import("./migration/20260703181610_event_created_column"), import("./migration/20260703190000_reset_v2_shell_event_payloads"), import("./migration/20260703200000_reset_v2_session_events"), + import("./migration/20260704161518_durable_session_inbox"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260704161518_durable_session_inbox.ts b/packages/core/src/database/migration/20260704161518_durable_session_inbox.ts new file mode 100644 index 000000000000..72c249095d27 --- /dev/null +++ b/packages/core/src/database/migration/20260704161518_durable_session_inbox.ts @@ -0,0 +1,43 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260704161518_durable_session_inbox", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`PRAGMA foreign_keys=OFF;`) + yield* tx.run(` + CREATE TABLE \`__new_session_input\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text NOT NULL, + \`type\` text NOT NULL, + \`prompt\` text, + \`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( + `INSERT INTO \`__new_session_input\`(\`id\`, \`session_id\`, \`type\`, \`prompt\`, \`delivery\`, \`admitted_seq\`, \`promoted_seq\`, \`time_created\`) SELECT \`id\`, \`session_id\`, 'prompt', \`prompt\`, \`delivery\`, \`admitted_seq\`, \`promoted_seq\`, \`time_created\` FROM \`session_input\`;`, + ) + yield* tx.run(`DROP TABLE \`session_input\`;`) + yield* tx.run(`ALTER TABLE \`__new_session_input\` RENAME TO \`session_input\`;`) + yield* tx.run(`PRAGMA foreign_keys=ON;`) + yield* tx.run(`DROP INDEX IF EXISTS \`session_input_session_pending_delivery_seq_idx\`;`) + yield* tx.run( + `CREATE INDEX \`session_input_session_pending_type_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`type\`,\`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\`);`, + ) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 17a7d12aedd8..a04353fd4d65 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -170,8 +170,9 @@ export default { CREATE TABLE \`session_input\` ( \`id\` text PRIMARY KEY, \`session_id\` text NOT NULL, - \`prompt\` text NOT NULL, - \`delivery\` text NOT NULL, + \`type\` text NOT NULL, + \`prompt\` text, + \`delivery\` text, \`admitted_seq\` integer NOT NULL, \`promoted_seq\` integer, \`time_created\` integer NOT NULL, @@ -259,7 +260,10 @@ 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\`);`, + `CREATE INDEX \`session_input_session_pending_type_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`type\`,\`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\`);`, diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 05d64c8efbff..bf08e4ffeb48 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -33,7 +33,6 @@ import { MessageDecodeError } from "./session/error" import { SessionEvent } from "./session/event" import { SessionInput } from "./session/input" import { Snapshot } from "./snapshot" -import { SessionCompaction } from "./session/compaction" import { SessionRevert } from "./session/revert" import { Revert } from "@opencode-ai/schema/revert" import { FSUtil } from "./fs-util" @@ -94,6 +93,7 @@ type CreateInput = CreateBaseInput & ({ location: Location.Ref; parentID?: never } | { parentID: SessionSchema.ID; location?: never }) type CompactInput = { + id?: SessionMessage.ID sessionID: SessionSchema.ID } @@ -119,6 +119,13 @@ export class PromptConflictError extends Schema.TaggedErrorClass()( + "Session.CompactionConflictError", + { + sessionID: SessionSchema.ID, + inputID: SessionMessage.ID, + }, +) {} export class BusyError extends Schema.TaggedErrorClass()("Session.BusyError", { sessionID: SessionSchema.ID, }) {} @@ -133,6 +140,7 @@ export type Error = | MessageDecodeError | OperationUnavailableError | PromptConflictError + | CompactionConflictError | BusyError | SkillNotFoundError | CommandV2.NotFoundError @@ -220,7 +228,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 @@ -623,19 +631,20 @@ const layer = Layer.effect( }) }), compact: Effect.fn("V2Session.compact")(function* (input) { - const session = yield* result.get(input.sessionID) - // TODO: admit manual compaction as durable pending work, like prompt input, instead of rejecting active sessions. - if ((yield* execution.active).has(input.sessionID)) return yield* new BusyError({ sessionID: input.sessionID }) - const context = yield* store.context(input.sessionID) - const compacted = yield* Effect.gen(function* () { - const compaction = yield* SessionCompaction.Service - return yield* compaction.compactManual({ session, messages: context }) + yield* result.get(input.sessionID) + const inputID = input.id ?? SessionMessage.ID.create() + const admitted = yield* SessionInput.admitCompaction(db, events, { + id: inputID, + sessionID: input.sessionID, }).pipe( - Effect.provide(locations.get(session.location)), - Effect.catch(() => Effect.succeed(false)), + Effect.catchDefect((defect) => + defect instanceof SessionInput.LifecycleConflict + ? new CompactionConflictError({ sessionID: input.sessionID, inputID }) + : Effect.die(defect), + ), ) - if (!compacted) return yield* new OperationUnavailableError({ operation: "compact" }) - return undefined + yield* execution.wake(input.sessionID) + return admitted }), wait: Effect.fn("V2Session.wait")(function* (sessionID) { yield* result.get(sessionID) diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 773fa476593b..2be346b6cf79 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -268,7 +268,9 @@ const make = (dependencies: Dependencies) => { if (context === undefined || context <= 0) return false const selected = select(input.messages, config.tokens) if (!selected) return false - const previousSummary = input.messages.find((message) => message.type === "compaction") + const previousSummary = input.messages.find( + (message) => message.type === "compaction" && message.status === "completed", + ) const hasHead = selected.head.length > 0 if (!hasHead && previousSummary?.type !== "compaction" && !input.force) return false const forcedShortContext = input.force && !hasHead diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index 9a6a75fe2f32..ae2f374bb455 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -1,4 +1,4 @@ -import { and, asc, desc, eq, gt, gte, ne, or } from "drizzle-orm" +import { and, asc, desc, eq, gt, gte, ne, or, sql } from "drizzle-orm" import { Effect, Schema } from "effect" import { Database } from "../database/database" import { MessageDecodeError } from "./error" @@ -14,7 +14,13 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService return yield* db .select({ seq: SessionMessageTable.seq }) .from(SessionMessageTable) - .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) + .where( + and( + eq(SessionMessageTable.session_id, sessionID), + eq(SessionMessageTable.type, "compaction"), + sql`json_extract(${SessionMessageTable.data}, '$.status') = 'completed'`, + ), + ) .orderBy(desc(SessionMessageTable.seq)) .limit(1) .get() diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index 1d0dfc71c3be..c5d6cb7cc3ab 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/input.ts @@ -2,9 +2,10 @@ export * as SessionInput from "./input" import { and, asc, eq, isNull } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" -import { Admitted, Delivery } from "@opencode-ai/schema/session-input" +import { Admitted, Compaction, Delivery, Entry } from "@opencode-ai/schema/session-input" import type { Database } from "../database/database" import type { EventV2 } from "../event" +import { KeyedMutex } from "../effect/keyed-mutex" import { SessionEvent } from "./event" import { SessionMessage } from "./message" import { Prompt } from "./prompt" @@ -13,30 +14,66 @@ import { SessionInputTable, SessionMessageTable } from "./sql" type DatabaseService = Database.Interface["db"] -export { Admitted, Delivery } +export { Admitted, Compaction, Delivery, Entry } const decodePrompt = Schema.decodeUnknownSync(Prompt) const encodePrompt = Schema.encodeSync(Prompt) +const inboxLocks = KeyedMutex.makeUnsafe() -const fromRow = (row: typeof SessionInputTable.$inferSelect): Admitted => - Admitted.make({ +export class LifecycleConflict extends Schema.TaggedErrorClass()("SessionInput.LifecycleConflict", { + id: SessionMessage.ID, +}) {} + +const fromRow = (row: typeof SessionInputTable.$inferSelect): Entry => { + 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.prompt || !row.delivery) throw new LifecycleConflict({ id: base.id }) + return Admitted.make({ + ...base, + type: "prompt", prompt: decodePrompt(row.prompt), delivery: row.delivery, - timeCreated: DateTime.makeUnsafe(row.time_created), ...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }), }) +} 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) return row === undefined ? undefined : fromRow(row) }) -export class LifecycleConflict extends Schema.TaggedErrorClass()("SessionInput.LifecycleConflict", { - id: SessionMessage.ID, -}) {} +export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(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)) + .limit(1) + .get() + .pipe(Effect.orDie) + if (!row) return + const entry = fromRow(row) + return entry.type === "compaction" ? entry : undefined +}) export const admit = Effect.fn("SessionInput.admit")(function* ( db: DatabaseService, @@ -49,7 +86,10 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( }, ) { const existing = yield* find(db, input.id) - if (existing !== undefined) return existing + if (existing !== undefined) { + if (existing.type !== "prompt") return yield* Effect.die(new LifecycleConflict({ id: input.id })) + return existing + } return yield* events .publish(SessionEvent.PromptAdmitted, { inputID: input.id, @@ -63,6 +103,7 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( ? Effect.die(new Error("Prompt admission event is missing aggregate sequence")) : Effect.succeed( Admitted.make({ + type: "prompt", admittedSeq: event.durable.seq, id: input.id, sessionID: input.sessionID, @@ -73,11 +114,52 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( ), ), Effect.catchDefect((defect) => - find(db, input.id).pipe(Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect)))), + find(db, input.id).pipe( + Effect.flatMap((stored) => (stored?.type === "prompt" ? Effect.succeed(stored) : Effect.die(defect))), + ), ), ) }) +export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(function* ( + db: DatabaseService, + events: EventV2.Interface, + input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID }, +) { + return yield* inboxLocks.withLock(input.sessionID)( + Effect.gen(function* () { + const exact = yield* find(db, input.id) + if (exact) { + 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) + if (pending) return pending + return yield* events + .publish(SessionEvent.Compaction.Admitted, { + inputID: input.id, + sessionID: input.sessionID, + }) + .pipe( + 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( + Effect.flatMap((stored) => + stored ? Effect.succeed(stored) : Effect.die(new LifecycleConflict({ id: input.id })), + ), + ) + }), + Effect.catchDefect((defect) => + pendingCompaction(db, input.sessionID).pipe( + Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))), + ), + ), + ) + }), + ) +}) + export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(function* ( db: DatabaseService, input: { @@ -101,6 +183,7 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio .values({ id: input.id, session_id: input.sessionID, + type: "prompt", admitted_seq: input.admittedSeq, prompt: encodePrompt(input.prompt), delivery: input.delivery, @@ -113,6 +196,44 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio if (!stored) return yield* Effect.die(new LifecycleConflict({ id: input.id })) }) +export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompactionAdmitted")(function* ( + db: DatabaseService, + input: { + readonly admittedSeq: number + readonly id: SessionMessage.ID + readonly sessionID: SessionSchema.ID + readonly timeCreated: DateTime.Utc + }, +) { + const message = yield* db + .select({ id: SessionMessageTable.id }) + .from(SessionMessageTable) + .where(eq(SessionMessageTable.id, input.id)) + .get() + .pipe(Effect.orDie) + if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: input.id })) + const stored = yield* db + .insert(SessionInputTable) + .values({ + id: input.id, + session_id: input.sessionID, + type: "compaction", + admitted_seq: input.admittedSeq, + time_created: DateTime.toEpochMillis(input.timeCreated), + }) + .onConflictDoNothing() + .returning() + .get() + .pipe(Effect.orDie) + if (stored) { + const entry = fromRow(stored) + return entry.type === "compaction" ? entry : yield* Effect.die(new LifecycleConflict({ id: entry.id })) + } + const pending = yield* pendingCompaction(db, input.sessionID) + if (pending) return pending + return yield* Effect.die(new LifecycleConflict({ id: input.id })) +}) + export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromoted")(function* ( db: DatabaseService, input: { @@ -121,6 +242,7 @@ export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromot 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 }) @@ -128,6 +250,7 @@ export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromot and( eq(SessionInputTable.id, input.id), eq(SessionInputTable.session_id, input.sessionID), + eq(SessionInputTable.type, "prompt"), isNull(SessionInputTable.promoted_seq), ), ) @@ -136,29 +259,58 @@ export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromot .pipe(Effect.orDie) if (updated) { const stored = fromRow(updated) - if (stored.sessionID !== input.sessionID) return yield* Effect.die(new LifecycleConflict({ id: input.id })) + if (stored.type !== "prompt" || stored.sessionID !== input.sessionID) + return yield* Effect.die(new LifecycleConflict({ id: input.id })) return stored } - - // Every PromptPromoted event is published from an admitted inbox row, so a missing or - // divergent row on replay is an invariant violation. const stored = yield* find(db, input.id) - if (!stored || stored.sessionID !== input.sessionID || stored.promotedSeq !== input.promotedSeq) + if ( + !stored || + stored.type !== "prompt" || + stored.sessionID !== input.sessionID || + stored.promotedSeq !== input.promotedSeq + ) return yield* Effect.die(new LifecycleConflict({ id: input.id })) return stored }) +export const settleCompaction = Effect.fn("SessionInput.settleCompaction")(function* ( + db: DatabaseService, + input: { readonly sessionID: SessionSchema.ID; readonly handledSeq: number }, +) { + 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), + ), + ) + .returning() + .get() + .pipe(Effect.orDie) + if (updated) { + const stored = fromRow(updated) + return stored.type === "compaction" ? stored : yield* Effect.die(new LifecycleConflict({ id: stored.id })) + } + return undefined +}) + export const hasPending = Effect.fn("SessionInput.hasPending")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, delivery: Delivery, ) { + if (yield* pendingCompaction(db, sessionID)) return false const row = yield* db .select({ id: SessionInputTable.id }) .from(SessionInputTable) .where( and( eq(SessionInputTable.session_id, sessionID), + eq(SessionInputTable.type, "prompt"), isNull(SessionInputTable.promoted_seq), eq(SessionInputTable.delivery, delivery), ), @@ -181,42 +333,44 @@ export const equivalent = ( input.sessionID === expected.sessionID && JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt)) -const matchesProjection = ( - input: Admitted, - expected: { - readonly sessionID: SessionSchema.ID - readonly prompt: Prompt - readonly delivery: Delivery - readonly timeCreated: DateTime.Utc - }, -) => - equivalent(input, expected) && - DateTime.toEpochMillis(input.timeCreated) === DateTime.toEpochMillis(expected.timeCreated) - const publish = Effect.fn("SessionInput.publish")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, rows: ReadonlyArray, ) { - for (const row of rows) { - const id = SessionMessage.ID.make(row.id) - yield* events - .publish(SessionEvent.PromptPromoted, { - sessionID, - inputID: id, - }) - .pipe( - Effect.catchDefect((defect) => - defect instanceof LifecycleConflict - ? find(db, id).pipe( - Effect.flatMap((stored) => (stored?.promotedSeq === undefined ? Effect.die(defect) : Effect.void)), - ) - : Effect.die(defect), - ), + return yield* inboxLocks.withLock(sessionID)( + Effect.gen(function* () { + if (yield* pendingCompaction(db, sessionID)) return 0 + yield* Effect.forEach( + rows, + (row) => { + const entry = fromRow(row) + if (entry.type !== "prompt") return Effect.die(new LifecycleConflict({ id: entry.id })) + return events + .publish(SessionEvent.PromptPromoted, { + sessionID, + inputID: entry.id, + }) + .pipe( + Effect.catchDefect((defect) => + defect instanceof LifecycleConflict + ? find(db, entry.id).pipe( + Effect.flatMap((stored) => + stored?.type === "prompt" && stored.promotedSeq !== undefined + ? Effect.void + : Effect.die(defect), + ), + ) + : Effect.die(defect), + ), + ) + }, + { discard: true }, ) - } - return rows.length + return rows.length + }), + ) }) export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( @@ -224,12 +378,14 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( events: EventV2.Interface, sessionID: SessionSchema.ID, ) { + if (yield* pendingCompaction(db, sessionID)) return 0 const rows = yield* db .select() .from(SessionInputTable) .where( and( eq(SessionInputTable.session_id, sessionID), + eq(SessionInputTable.type, "prompt"), isNull(SessionInputTable.promoted_seq), eq(SessionInputTable.delivery, "steer"), ), @@ -245,12 +401,14 @@ export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(fun events: EventV2.Interface, sessionID: SessionSchema.ID, ) { + if (yield* pendingCompaction(db, sessionID)) return false const row = yield* db .select() .from(SessionInputTable) .where( and( eq(SessionInputTable.session_id, sessionID), + eq(SessionInputTable.type, "prompt"), isNull(SessionInputTable.promoted_seq), eq(SessionInputTable.delivery, "queue"), ), diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 0fc9b75f6a8e..affd5441a9db 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -16,8 +16,10 @@ export interface Adapter { readonly getShell: ( shellID: SessionMessage.Shell["shell"]["id"], ) => Effect.Effect + readonly getCompaction: () => Effect.Effect readonly updateAssistant: (assistant: SessionMessage.Assistant) => Effect.Effect readonly updateShell: (shell: SessionMessage.Shell) => Effect.Effect + readonly updateCompaction: (compaction: SessionMessage.Compaction) => Effect.Effect readonly appendMessage: (message: SessionMessage.Message) => Effect.Effect } @@ -26,6 +28,10 @@ export function memory(state: MemoryState): Adapter { state.messages.findLastIndex((message) => message.id === messageID) const shellIndex = (messageID: SessionMessage.ID) => state.messages.findLastIndex((message) => message.id === messageID) + const compactionIndex = () => + state.messages.findLastIndex( + (message) => message.type === "compaction" && (message.status === "queued" || message.status === "running"), + ) // A newer step supersedes stale incomplete rows; never resume an older assistant projection. const latestAssistantIndex = () => state.messages.findLastIndex((message) => message.type === "assistant") @@ -62,6 +68,13 @@ export function memory(state: MemoryState): Adapter { }) }) }, + getCompaction() { + return Effect.sync(() => { + const index = compactionIndex() + const message = state.messages[index] + return message?.type === "compaction" ? message : undefined + }) + }, updateAssistant(assistant) { return Effect.sync(() => { const index = assistantIndex(assistant.id) @@ -80,6 +93,12 @@ export function memory(state: MemoryState): Adapter { state.messages[index] = shell }) }, + updateCompaction(compaction) { + return Effect.sync(() => { + const index = state.messages.findLastIndex((message) => message.id === compaction.id) + if (index >= 0) state.messages[index] = compaction + }) + }, appendMessage(message) { return Effect.sync(() => { state.messages.push(message) @@ -423,21 +442,60 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { } }) }, - "session.compaction.started": () => Effect.void, - "session.compaction.delta": () => Effect.void, - "session.compaction.ended": (event) => { - return adapter.appendMessage( + "session.compaction.admitted": (event) => + adapter.appendMessage( SessionMessage.Compaction.make({ - id: SessionMessage.ID.fromEvent(event.id), + id: event.data.inputID, type: "compaction", + status: "queued", metadata: event.metadata, - reason: event.data.reason, - summary: event.data.text, - recent: event.data.recent, + reason: "manual", + summary: "", + recent: "", time: { created: event.created }, }), - ) + ), + "session.compaction.started": (event) => + Effect.gen(function* () { + if (event.data.reason !== "manual") return + const current = yield* adapter.getCompaction() + if (!current) return + yield* adapter.updateCompaction({ ...current, status: "running" }) + }), + "session.compaction.delta": () => Effect.void, + "session.compaction.ended": (event) => { + return Effect.gen(function* () { + const current = event.data.reason === "manual" ? yield* adapter.getCompaction() : undefined + if (current) { + yield* adapter.updateCompaction({ + ...current, + status: "completed", + reason: event.data.reason, + summary: event.data.text, + recent: event.data.recent, + }) + return + } + yield* adapter.appendMessage( + SessionMessage.Compaction.make({ + id: SessionMessage.ID.fromEvent(event.id), + type: "compaction", + status: "completed", + metadata: event.metadata, + reason: event.data.reason, + summary: event.data.text, + recent: event.data.recent, + time: { created: event.created }, + }), + ) + }) }, + "session.compaction.failed": () => + Effect.gen(function* () { + const current = yield* adapter.getCompaction() + if (!current) return + yield* adapter.updateCompaction({ ...current, status: "failed" }) + }), "session.revert.staged": () => Effect.void, "session.revert.cleared": () => Effect.void, "session.revert.committed": () => Effect.void, diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 92c3b32c0708..e638574ea29f 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -243,6 +243,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( eq(SessionMessageTable.session_id, event.data.parentID), gt(SessionMessageTable.seq, cursor), copiedSeq === 0 ? undefined : lt(SessionMessageTable.seq, copiedSeq + 1), + sql`${SessionMessageTable.type} != 'compaction' or json_extract(${SessionMessageTable.data}, '$.status') not in ('queued', 'running')`, ), ) .orderBy(asc(SessionMessageTable.seq)) @@ -292,11 +293,12 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( .values( inputRows.flatMap((row) => { const id = idMap.get(row.id) - return id + return id && row.type === "prompt" ? [ { id, session_id: event.data.sessionID, + type: "prompt" as const, prompt: row.prompt, delivery: row.delivery, admitted_seq: row.admitted_seq, @@ -426,8 +428,30 @@ function run(db: DatabaseService, event: MessageEvent) { return message.type === "shell" ? message : undefined }) }, + getCompaction() { + return Effect.gen(function* () { + const row = yield* db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, event.data.sessionID), + eq(SessionMessageTable.type, "compaction"), + sql`json_extract(${SessionMessageTable.data}, '$.status') in ('queued', 'running')`, + ), + ) + .orderBy(desc(SessionMessageTable.seq)) + .limit(1) + .get() + .pipe(Effect.orDie) + if (!row) return + const message = decodeRow(row) + return message.type === "compaction" ? message : undefined + }) + }, updateAssistant: updateMessage, updateShell: updateMessage, + updateCompaction: updateMessage, appendMessage, } yield* SessionMessageUpdater.update(adapter, event) @@ -634,6 +658,20 @@ const layer = Layer.effectDiscard( }) }), ) + yield* events.project(SessionEvent.Compaction.Admitted, (event) => + Effect.gen(function* () { + if (event.durable === undefined) + return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) + const admitted = yield* SessionInput.projectCompactionAdmitted(db, { + admittedSeq: event.durable.seq, + id: event.data.inputID, + sessionID: event.data.sessionID, + timeCreated: event.created, + }) + if (admitted.id !== event.data.inputID) return + yield* run(db, event) + }), + ) yield* events.project(SessionEvent.Execution.Succeeded, (event) => run(db, event)) yield* events.project(SessionEvent.Execution.Failed, (event) => run(db, event)) yield* events.project(SessionEvent.Execution.Interrupted, (event) => run(db, event)) @@ -664,7 +702,30 @@ const layer = Layer.effectDiscard( yield* events.project(SessionEvent.Reasoning.Started, (event) => run(db, event)) yield* events.project(SessionEvent.Reasoning.Ended, (event) => run(db, event)) yield* events.project(SessionEvent.RetryScheduled, (event) => run(db, event)) - yield* events.project(SessionEvent.Compaction.Ended, (event) => run(db, event)) + yield* events.project(SessionEvent.Compaction.Started, (event) => run(db, event)) + yield* events.project(SessionEvent.Compaction.Ended, (event) => + Effect.gen(function* () { + yield* run(db, event) + 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* events.project(SessionEvent.Compaction.Failed, (event) => + Effect.gen(function* () { + yield* run(db, event) + if (event.durable === undefined) + return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) + yield* SessionInput.settleCompaction(db, { + sessionID: event.data.sessionID, + handledSeq: event.durable.seq, + }) + }), + ) yield* events.project(SessionEvent.RevertEvent.Staged, (event) => db .update(SessionTable) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 56011cb90626..55d318612135 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -220,7 +220,10 @@ const layer = Layer.effect( toolChoice: isLastStep ? "none" : undefined, }) // Automatic compaction completed; rebuild the request from compacted history. - if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) + if ( + !(yield* SessionInput.pendingCompaction(db, session.id)) && + (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) + ) return { _tag: "RestartAfterCompaction", step: currentStep } as const const startSnapshot = yield* snapshots.capture() const publisher = createLLMEventPublisher(events, { @@ -503,11 +506,36 @@ const layer = Layer.effect( } }) + const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( + sessionID: SessionSchema.ID, + ) { + const pending = yield* SessionInput.pendingCompaction(db, sessionID) + if (!pending) return false + const session = yield* getSession(sessionID) + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const compacted = yield* restore( + Effect.gen(function* () { + return yield* compaction.compactManual({ + session, + messages: yield* store.context(sessionID), + }) + }), + ).pipe(Effect.exit) + if (Exit.isSuccess(compacted) && compacted.value) return true + yield* events.publish(SessionEvent.Compaction.Failed, { sessionID }) + if (Exit.isFailure(compacted)) return yield* Effect.failCause(compacted.cause) + return true + }), + ) + }) + // Execution lifecycle is published per busy period by SessionExecution, not per drain here. const drain = Effect.fn("SessionRunner.drain")(function* (input: { readonly sessionID: SessionSchema.ID 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") if (!input.force && !hasSteer && !hasQueue) return @@ -531,11 +559,19 @@ const layer = Layer.effect( } needsContinuation = result.needsContinuation step = result.step + 1 + if (needsContinuation) { + promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer" + continue + } + yield* runPendingCompaction(input.sessionID) promotion = "steer" - if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") + needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") } - shouldRun = yield* SessionInput.hasPending(db, input.sessionID, "queue") - promotion = shouldRun ? "queue" : undefined + yield* runPendingCompaction(input.sessionID) + const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") + const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") + shouldRun = hasSteer || hasQueue + promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined } }) diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index cb83c0fdb2bf..35713eabe639 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -161,6 +161,7 @@ function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] case "assistant": return assistant(message, model) case "compaction": + if (message.status !== "completed") return [] return [ Message.make({ id: message.id, diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 739d3f188804..58ff1b641ffc 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -1,4 +1,5 @@ import { sqliteTable, text, integer, index, primaryKey, real, uniqueIndex } from "drizzle-orm/sqlite-core" +import { sql } from "drizzle-orm" import { directoryColumn, pathColumn } from "../database/path" import { ProjectTable } from "../project/sql" import type { SessionMessage } from "./message" @@ -146,8 +147,9 @@ export const SessionInputTable = sqliteTable( .$type() .notNull() .references(() => SessionTable.id, { onDelete: "cascade" }), - prompt: text({ mode: "json" }).notNull().$type(), - delivery: text().$type().notNull(), + type: text().$type().notNull(), + prompt: text({ mode: "json" }).$type(), + delivery: text().$type(), admitted_seq: integer().notNull(), promoted_seq: integer(), time_created: integer() @@ -155,12 +157,16 @@ export const SessionInputTable = sqliteTable( .$default(() => Date.now()), }, (table) => [ - index("session_input_session_pending_delivery_seq_idx").on( + index("session_input_session_pending_type_delivery_seq_idx").on( table.session_id, table.promoted_seq, + table.type, table.delivery, table.admitted_seq, ), + uniqueIndex("session_input_session_pending_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), ], diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index b190170f2333..dafe6193b2a1 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -16,6 +16,7 @@ import contextEpochAgentMigration from "@opencode-ai/core/database/migration/202 import simplifyIntegrationCredentialsMigration from "@opencode-ai/core/database/migration/20260611192811_lush_chimera" import simplifySessionInputMigration 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/20260704161518_durable_session_inbox" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" @@ -107,13 +108,14 @@ 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_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_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_delivery_seq_idx" }, + { name: "session_input_session_pending_compaction_idx" }, + { name: "session_input_session_pending_type_delivery_seq_idx" }, { name: "session_input_session_promoted_seq_idx" }, { name: "session_message_session_seq_idx" }, { name: "session_message_session_time_created_id_idx" }, @@ -282,7 +284,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, prompt, delivery, admitted_seq, time_created) VALUES ('input', 'session', '{}', 'steer', 9, 1)`, + sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'prompt', '{}', '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, '{}')`, @@ -343,6 +345,37 @@ describe("DatabaseMigration", () => { ) }) + test("preserves admitted prompts while generalizing the durable inbox", async () => { + await run( + Effect.gen(function* () { + const db = yield* makeDb + yield* db.run( + sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, prompt text NOT NULL, delivery text NOT NULL, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`, + ) + yield* db.run( + sql`INSERT INTO session_input (id, session_id, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('input', 'session', '{"text":"hello"}', 'steer', 4, NULL, 1)`, + ) + + yield* DatabaseMigration.applyOnly(db, [durableSessionInboxMigration]) + + expect( + yield* db.all( + sql`SELECT id, type, prompt, delivery, admitted_seq, promoted_seq FROM session_input ORDER BY admitted_seq`, + ), + ).toEqual([ + { + id: "input", + type: "prompt", + prompt: '{"text":"hello"}', + delivery: "steer", + admitted_seq: 4, + promoted_seq: null, + }, + ]) + }), + ) + }) + 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 4a9a520d59df..e482239aa7f2 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -74,7 +74,7 @@ const it = testEffect( ) describe("SessionV2.compact", () => { - it.effect("manually compacts the active session context", () => + it.effect("durably admits and coalesces manual compaction", () => Effect.gen(function* () { requests = [] const session = yield* SessionV2.Service @@ -94,13 +94,22 @@ describe("SessionV2.compact", () => { inputID: messageID, }) - yield* session.compact({ sessionID: created.id }) + expect(yield* session.compact({ id: messageID, sessionID: created.id }).pipe(Effect.flip)).toMatchObject({ + _tag: "Session.CompactionConflictError", + inputID: messageID, + }) + const first = yield* session.compact({ sessionID: created.id }) + const second = yield* session.compact({ sessionID: created.id }) - expect(requests).toHaveLength(1) - expect(JSON.stringify(requests[0]?.messages)).toContain("Please compact this session history.") - expect(yield* session.context(created.id)).toMatchObject([ - { type: "compaction", reason: "manual", summary: "manual session summary", recent: "" }, - ]) + expect(second.id).toBe(first.id) + expect(requests).toHaveLength(0) + expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toMatchObject({ + type: "compaction", + status: "queued", + reason: "manual", + summary: "", + recent: "", + }) }), ) }) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index e7d3f5189f82..80773cc6e85a 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -233,7 +233,9 @@ describe("SessionV2.prompt", () => { expect(message.prompt.files).toEqual([ { uri: "data:image/png;base64,aGVsbG8=", name: "image.png", mime: "image/png" }, ]) - expect((yield* admitted(message.id))?.prompt.files).toEqual(message.prompt.files) + const stored = yield* admitted(message.id) + expect(stored?.type).toBe("prompt") + if (stored?.type === "prompt") expect(stored.prompt.files).toEqual(message.prompt.files) }), ) diff --git a/packages/core/test/session-runner-message.test.ts b/packages/core/test/session-runner-message.test.ts index 4595e48f285a..99cd47903369 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -101,8 +101,9 @@ describe("toLLMMessages", () => { }), SessionMessage.Compaction.make({ id: id("compaction"), - type: "compaction", - reason: "auto", + type: "compaction", + status: "completed", + reason: "auto", summary: "Earlier work", recent: "Recent work", time: { created }, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 054d61c0dcb8..b5fff0ac9604 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1316,6 +1316,97 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("runs one durable compaction barrier before later steer and queued prompts", () => + Effect.gen(function* () { + yield* setup + requests.length = 0 + currentModel = recoveryModel + const session = yield* SessionV2.Service + streamGate = yield* Deferred.make() + streamStarted = yield* Deferred.make() + responses = [ + fragmentFixture("text", "text-active", ["Active complete"]).completeEvents, + [LLMEvent.textDelta({ id: "summary", text: "durable summary" })], + fragmentFixture("text", "text-steer", ["Steer complete"]).completeEvents, + fragmentFixture("text", "text-queue", ["Queue complete"]).completeEvents, + ] + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Active work" }), resume: false }) + const active = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* Deferred.await(streamStarted) + + 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({ + id: first.id, + }) + expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({ + type: "compaction", + status: "queued", + }) + + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Steer after compaction" }), resume: false }) + yield* session.prompt({ + sessionID, + prompt: Prompt.make({ text: "Queue after compaction" }), + delivery: "queue", + resume: false, + }) + expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false) + + yield* Deferred.succeed(streamGate, undefined) + yield* Fiber.join(active) + + expect(requests).toHaveLength(4) + expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary") + expect(userTexts(requests[2])).toContain("Steer after compaction") + expect(userTexts(requests[3])).toContain("Queue after compaction") + expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({ + type: "compaction", + status: "completed", + summary: "durable summary", + }) + }), + ) + + it.effect("releases queued prompts when durable compaction fails", () => + Effect.gen(function* () { + yield* setup + requests.length = 0 + currentModel = recoveryModel + const session = yield* SessionV2.Service + streamGate = yield* Deferred.make() + streamStarted = yield* Deferred.make() + responses = [ + fragmentFixture("text", "text-active-failure", ["Active complete"]).completeEvents, + [], + fragmentFixture("text", "text-after-failure", ["Continued"]).completeEvents, + ] + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Active work" }), resume: false }) + const active = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* Deferred.await(streamStarted) + + const compaction = yield* session.compact({ sessionID }) + yield* session.prompt({ + sessionID, + prompt: Prompt.make({ text: "Continue after failure" }), + delivery: "queue", + resume: false, + }) + yield* Deferred.succeed(streamGate, undefined) + yield* Fiber.join(active) + + expect(requests).toHaveLength(3) + expect(userTexts(requests[2])).toContain("Continue after failure") + expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined() + expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ + type: "compaction", + status: "failed", + }) + }), + ) + it.effect("automatically compacts into a completed summary and retained recent turn", () => Effect.gen(function* () { yield* setup diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 527dcc6f6189..d5f3e4cce025 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -372,15 +372,16 @@ export const makeSessionGroup = (sessionLo .add( HttpApiEndpoint.post("session.compact", "/api/session/:sessionID/compact", { params: { sessionID: Session.ID }, - success: HttpApiSchema.NoContent, - error: [SessionNotFoundError, SessionBusyError, ServiceUnavailableError, UnknownError], + payload: Schema.Struct({ id: SessionMessage.ID.pipe(Schema.optional) }), + success: Schema.Struct({ data: SessionInput.Compaction }), + error: [ConflictError, SessionNotFoundError], }) .middleware(sessionLocationMiddleware) .annotateMerge( OpenApi.annotations({ identifier: "v2.session.compact", summary: "Compact session", - description: "Compact a session conversation.", + description: "Queue a durable session compaction request.", }), ), ) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 73f582b50162..e0433da78d71 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -426,6 +426,16 @@ export const RetryScheduled = Event.durable({ export type RetryScheduled = typeof RetryScheduled.Type export namespace Compaction { + export const Admitted = Event.durable({ + type: "session.compaction.admitted", + ...options, + schema: { + ...Base, + inputID: SessionMessage.ID, + }, + }) + export type Admitted = typeof Admitted.Type + export const Started = Event.durable({ type: "session.compaction.started", ...options, @@ -456,6 +466,13 @@ export namespace Compaction { }, }) export type Ended = typeof Ended.Type + + export const Failed = Event.durable({ + type: "session.compaction.failed", + ...options, + schema: Base, + }) + export type Failed = typeof Failed.Type } export namespace RevertEvent { @@ -506,9 +523,11 @@ export const Definitions = Event.inventory( Tool.Success, Tool.Failed, RetryScheduled, + Compaction.Admitted, Compaction.Started, Compaction.Delta, Compaction.Ended, + Compaction.Failed, RevertEvent.Staged, RevertEvent.Cleared, RevertEvent.Committed, diff --git a/packages/schema/src/session-input.ts b/packages/schema/src/session-input.ts index eefe68be2b67..a3e691dcd165 100644 --- a/packages/schema/src/session-input.ts +++ b/packages/schema/src/session-input.ts @@ -13,6 +13,7 @@ export type Delivery = SessionDelivery.Delivery export interface Admitted extends Schema.Schema.Type {} export const Admitted = Schema.Struct({ + type: Schema.Literal("prompt"), admittedSeq: NonNegativeInt, id: SessionMessage.ID, sessionID: SessionID, @@ -21,3 +22,16 @@ export const Admitted = Schema.Struct({ timeCreated: DateTimeUtcFromMillis, promotedSeq: NonNegativeInt.pipe(optional), }).annotate({ identifier: "SessionInput.Admitted" }) + +export interface Compaction extends Schema.Schema.Type {} +export const Compaction = Schema.Struct({ + type: Schema.Literal("compaction"), + admittedSeq: NonNegativeInt, + id: SessionMessage.ID, + sessionID: SessionID, + timeCreated: DateTimeUtcFromMillis, + handledSeq: NonNegativeInt.pipe(optional), +}).annotate({ identifier: "SessionInput.Compaction" }) + +export const Entry = Schema.Union([Admitted, Compaction]).pipe(Schema.toTaggedUnion("type")) +export type Entry = typeof Entry.Type diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index aafef1dba2a3..62e1a1038d3c 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -210,6 +210,7 @@ export const Assistant = Schema.Struct({ export interface Compaction extends Schema.Schema.Type {} export const Compaction = Schema.Struct({ type: Schema.Literal("compaction"), + status: Schema.Literals(["queued", "running", "completed", "failed"]), reason: Schema.Literals(["auto", "manual"]), summary: Schema.String, recent: Schema.String, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 946a75947b16..ead61c5d429c 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -124,8 +124,10 @@ describe("public event manifest", () => { "session.reasoning.started.1", "session.reasoning.ended.1", "session.retry.scheduled.1", + "session.compaction.admitted.1", "session.compaction.started.1", "session.compaction.ended.1", + "session.compaction.failed.1", "session.revert.staged.1", "session.revert.cleared.1", "session.revert.committed.1", diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 1a9003cacf5b..859e8859e823 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -6282,19 +6282,35 @@ export class Session3 extends HeyApiClient { /** * Compact session * - * Compact a session conversation. + * Queue a durable session compaction request. */ public compact( parameters: { sessionID: string + id?: string | null }, options?: Options, ) { - const params = buildClientParams([parameters], [{ args: [{ in: "path", key: "sessionID" }] }]) + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "path", key: "sessionID" }, + { in: "body", key: "id" }, + ], + }, + ], + ) return (options?.client ?? this.client).post({ url: "/api/session/{sessionID}/compact", ...options, ...params, + headers: { + "Content-Type": "application/json", + ...options?.headers, + ...params.headers, + }, }) } diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 189f0ab800f2..2cfd67fe3906 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -50,9 +50,11 @@ export type Event = | EventSessionToolSuccess | EventSessionToolFailed | EventSessionRetryScheduled + | EventSessionCompactionAdmitted | EventSessionCompactionStarted | EventSessionCompactionDelta | EventSessionCompactionEnded + | EventSessionCompactionFailed | EventSessionRevertStaged | EventSessionRevertCleared | EventSessionRevertCommitted @@ -1200,6 +1202,14 @@ export type GlobalEvent = { error: SessionStructuredError } } + | { + id: string + type: "session.compaction.admitted" + properties: { + sessionID: string + inputID: string + } + } | { id: string type: "session.compaction.started" @@ -1226,6 +1236,13 @@ export type GlobalEvent = { recent: string } } + | { + id: string + type: "session.compaction.failed" + properties: { + sessionID: string + } + } | { id: string type: "session.revert.staged" @@ -1765,8 +1782,10 @@ export type GlobalEvent = { | SyncEventSessionToolSuccess | SyncEventSessionToolFailed | SyncEventSessionRetryScheduled + | SyncEventSessionCompactionAdmitted | SyncEventSessionCompactionStarted | SyncEventSessionCompactionEnded + | SyncEventSessionCompactionFailed | SyncEventSessionRevertStaged | SyncEventSessionRevertCleared | SyncEventSessionRevertCommitted @@ -2937,8 +2956,10 @@ export type SessionDurableEvent = | SessionToolSuccess | SessionToolFailed | SessionRetryScheduled + | SessionCompactionAdmitted | SessionCompactionStarted | SessionCompactionEnded + | SessionCompactionFailed | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted @@ -3085,9 +3106,11 @@ export type V2Event = | SessionToolSuccess | SessionToolFailed | SessionRetryScheduled + | SessionCompactionAdmitted | SessionCompactionStarted | SessionCompactionDelta | SessionCompactionEnded + | SessionCompactionFailed | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted @@ -4154,6 +4177,21 @@ export type SyncEventSessionRetryScheduled = { } } +export type SyncEventSessionCompactionAdmitted = { + type: "sync" + id: string + syncEvent: { + type: "session.compaction.admitted.1" + id: string + seq: number + aggregateID: string + data: { + sessionID: string + inputID: string + } + } +} + export type SyncEventSessionCompactionStarted = { type: "sync" id: string @@ -4186,6 +4224,20 @@ export type SyncEventSessionCompactionEnded = { } } +export type SyncEventSessionCompactionFailed = { + type: "sync" + id: string + syncEvent: { + type: "session.compaction.failed.1" + id: string + seq: number + aggregateID: string + data: { + sessionID: string + } + } +} + export type SyncEventSessionRevertStaged = { type: "sync" id: string @@ -4347,6 +4399,7 @@ export type PromptInputFileAttachment = { } export type SessionInputAdmitted = { + type: "prompt" admittedSeq: number id: string sessionID: string @@ -4356,6 +4409,15 @@ export type SessionInputAdmitted = { promotedSeq?: number } +export type SessionInputCompaction = { + type: "compaction" + admittedSeq: number + id: string + sessionID: string + timeCreated: number + handledSeq?: number +} + export type SessionMessageAgentSelected = { id: string metadata?: { @@ -4572,6 +4634,7 @@ export type SessionMessageAssistant = { export type SessionMessageCompaction = { type: "compaction" + status: "queued" | "running" | "completed" | "failed" reason: "auto" | "manual" summary: string recent: string @@ -5242,6 +5305,25 @@ export type SessionRetryScheduled = { } } +export type SessionCompactionAdmitted = { + id: string + created: number + metadata?: { + [key: string]: unknown + } + type: "session.compaction.admitted" + durable: { + aggregateID: string + seq: number + version: number + } + location?: LocationRef + data: { + sessionID: string + inputID: string + } +} + export type SessionCompactionStarted = { id: string created: number @@ -5282,6 +5364,24 @@ export type SessionCompactionEnded = { } } +export type SessionCompactionFailed = { + id: string + created: number + metadata?: { + [key: string]: unknown + } + type: "session.compaction.failed" + durable: { + aggregateID: string + seq: number + version: number + } + location?: LocationRef + data: { + sessionID: string + } +} + export type SessionRevertStaged = { id: string created: number @@ -7258,6 +7358,15 @@ export type EventSessionRetryScheduled = { } } +export type EventSessionCompactionAdmitted = { + id: string + type: "session.compaction.admitted" + properties: { + sessionID: string + inputID: string + } +} + export type EventSessionCompactionStarted = { id: string type: "session.compaction.started" @@ -7287,6 +7396,14 @@ export type EventSessionCompactionEnded = { } } +export type EventSessionCompactionFailed = { + id: string + type: "session.compaction.failed" + properties: { + sessionID: string + } +} + export type EventSessionRevertStaged = { id: string type: "session.revert.staged" @@ -8450,9 +8567,11 @@ export type V2EventV2 = | SessionToolSuccessV2 | SessionToolFailedV2 | SessionRetryScheduledV2 + | SessionCompactionAdmittedV2 | SessionCompactionStartedV2 | SessionCompactionDeltaV2 | SessionCompactionEndedV2 + | SessionCompactionFailedV2 | SessionRevertStagedV2 | SessionRevertClearedV2 | SessionRevertCommittedV2 @@ -8578,6 +8697,7 @@ export type SessionV2InfoV2 = { } export type SessionInputAdmittedV2 = { + type: "prompt" admittedSeq: number id: string sessionID: string @@ -8587,6 +8707,15 @@ export type SessionInputAdmittedV2 = { promotedSeq?: number } +export type SessionInputCompactionV2 = { + type: "compaction" + admittedSeq: number + id: string + sessionID: string + timeCreated: number + handledSeq?: number +} + export type SessionMessageAgentSelectedV2 = { id: string metadata?: { @@ -8725,6 +8854,7 @@ export type SessionMessageAssistantV2 = { export type SessionMessageCompactionV2 = { type: "compaction" + status: "queued" | "running" | "completed" | "failed" reason: "auto" | "manual" summary: string recent: string @@ -9402,6 +9532,25 @@ export type SessionRetryScheduledV2 = { } } +export type SessionCompactionAdmittedV2 = { + id: string + created: number + metadata?: { + [key: string]: unknown + } + type: "session.compaction.admitted" + durable: { + aggregateID: string + seq: number + version: number + } + location?: LocationRefV2 + data: { + sessionID: string + inputID: string + } +} + export type SessionCompactionStartedV2 = { id: string created: number @@ -9442,6 +9591,24 @@ export type SessionCompactionEndedV2 = { } } +export type SessionCompactionFailedV2 = { + id: string + created: number + metadata?: { + [key: string]: unknown + } + type: "session.compaction.failed" + durable: { + aggregateID: string + seq: number + version: number + } + location?: LocationRefV2 + data: { + sessionID: string + } +} + export type SessionRevertStagedV2 = { id: string created: number @@ -15371,7 +15538,9 @@ export type V2SessionShellResponses = { export type V2SessionShellResponse = V2SessionShellResponses[keyof V2SessionShellResponses] export type V2SessionCompactData = { - body?: never + body: { + id?: string | null + } path: { sessionID: string } @@ -15393,26 +15562,20 @@ export type V2SessionCompactErrors = { */ 404: SessionNotFoundError /** - * SessionBusyError - */ - 409: SessionBusyError - /** - * UnknownError - */ - 500: UnknownErrorV2 - /** - * ServiceUnavailableError + * ConflictError */ - 503: ServiceUnavailableErrorV2 + 409: ConflictErrorV2 } export type V2SessionCompactError = V2SessionCompactErrors[keyof V2SessionCompactErrors] export type V2SessionCompactResponses = { /** - * + * Success */ - 204: void + 200: { + data: SessionInputCompactionV2 + } } export type V2SessionCompactResponse = V2SessionCompactResponses[keyof V2SessionCompactResponses] diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 8a3cf1b13053..03ce799aa3b6 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -344,44 +344,26 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.compact", Effect.fn(function* (ctx) { - yield* session.compact({ sessionID: ctx.params.sessionID }).pipe( - Effect.catchTag("Session.NotFoundError", (error) => - Effect.fail( - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), + return { + data: yield* session.compact({ sessionID: ctx.params.sessionID, id: ctx.payload.id }).pipe( + Effect.catchTag("Session.NotFoundError", (error) => + Effect.fail( + new SessionNotFoundError({ + sessionID: error.sessionID, + message: `Session not found: ${error.sessionID}`, + }), + ), ), - ), - Effect.catchTag("Session.OperationUnavailableError", (error) => - Effect.fail( - new ServiceUnavailableError({ - message: `Session ${error.operation} is not available yet`, - service: `session.${error.operation}`, - }), + Effect.catchTag("Session.CompactionConflictError", (error) => + Effect.fail( + new ConflictError({ + message: `Compaction input ID conflicts with an existing durable record: ${error.inputID}`, + resource: error.inputID, + }), + ), ), ), - Effect.catchTag( - "Session.BusyError", - (error) => - new SessionBusyError({ - sessionID: error.sessionID, - message: `Session is busy: ${error.sessionID}`, - }), - ), - Effect.catchTag("Session.MessageDecodeError", (error) => { - const ref = `err_${crypto.randomUUID().slice(0, 8)}` - return Effect.logError("failed to decode session message during compaction").pipe( - Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }), - Effect.andThen( - Effect.fail( - new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }), - ), - ), - ) - }), - ) - return HttpApiSchema.NoContent.make() + } }), ) .handle( diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 878c1fcd1578..3e23659e6742 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -52,6 +52,7 @@ type Data = { family: Record status: Record compaction: Partial> + compactionReason: Partial> message: Record permission: Record question: Record @@ -87,6 +88,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ family: {}, status: {}, compaction: {}, + compactionReason: {}, message: {}, permission: {}, question: {}, @@ -139,6 +141,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const item = messages.findLast((item) => item.type === "shell" && item.shell.id === shellID) return item?.type === "shell" ? item : undefined }, + compaction(messages: SessionMessage[]) { + const item = messages.findLast( + (item) => item.type === "compaction" && (item.status === "queued" || item.status === "running"), + ) + return item?.type === "compaction" ? item : undefined + }, latestTool(assistant: SessionMessageAssistant | undefined, callID?: string) { return assistant?.content.findLast( (item): item is SessionMessageAssistantTool => @@ -542,8 +550,28 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ case "session.execution.started": setSessionStatus(event.data.sessionID, "running") break + case "session.compaction.admitted": + message.update(event.data.sessionID, (draft, index) => { + if (message.compaction(draft)) return + message.append(draft, index, { + id: event.data.inputID, + type: "compaction", + status: "queued", + reason: "manual", + summary: "", + recent: "", + time: { created: event.created }, + }) + }) + break case "session.compaction.started": setStore("session", "compaction", event.data.sessionID, "") + setStore("session", "compactionReason", event.data.sessionID, event.data.reason) + if (event.data.reason === "manual") + message.update(event.data.sessionID, (draft) => { + const current = message.compaction(draft) + if (current) current.status = "running" + }) break case "session.execution.succeeded": case "session.execution.failed": @@ -551,6 +579,8 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setSessionStatus(event.data.sessionID, "idle") if (store.session.compaction[event.data.sessionID] !== undefined) setStore("session", "compaction", event.data.sessionID, undefined) + if (store.session.compactionReason[event.data.sessionID] !== undefined) + setStore("session", "compactionReason", event.data.sessionID, undefined) message.update(event.data.sessionID, (draft) => { const currentAssistant = message.activeAssistant(draft) if (currentAssistant) currentAssistant.retry = undefined @@ -567,13 +597,28 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ break case "session.compaction.delta": setStore("session", "compaction", event.data.sessionID, (text) => (text ?? "") + event.data.text) + if (store.session.compactionReason[event.data.sessionID] === "manual") + message.update(event.data.sessionID, (draft) => { + const current = message.compaction(draft) + if (current) current.summary += event.data.text + }) break case "session.compaction.ended": setStore("session", "compaction", event.data.sessionID, undefined) + setStore("session", "compactionReason", event.data.sessionID, undefined) message.update(event.data.sessionID, (draft, index) => { + const current = event.data.reason === "manual" ? message.compaction(draft) : undefined + if (current) { + current.status = "completed" + current.reason = event.data.reason + current.summary = event.data.text + current.recent = event.data.recent + return + } message.append(draft, index, { id: messageIDFromEvent(event.id), type: "compaction", + status: "completed", reason: event.data.reason, summary: event.data.text, recent: event.data.recent, @@ -581,6 +626,14 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) }) break + case "session.compaction.failed": + setStore("session", "compaction", event.data.sessionID, undefined) + setStore("session", "compactionReason", event.data.sessionID, undefined) + message.update(event.data.sessionID, (draft) => { + const current = message.compaction(draft) + if (current) current.status = "failed" + }) + break case "permission.v2.asked": if (store.session.permission[event.data.sessionID]?.some((request) => request.id === event.data.id)) break setStore("session", "permission", event.data.sessionID, [ diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index a959882ebda0..5411368e8e97 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -171,6 +171,16 @@ export function Session() { }) onCleanup(() => setEpilogue()) const messages = sessionMessages + const transientCompaction = createMemo(() => { + if ( + messages().some( + (message) => message.type === "compaction" && (message.status === "queued" || message.status === "running"), + ) + ) + return + const text = data.session.compaction(route.sessionID) + return text === undefined ? undefined : { text } + }) const descendantSessionIDs = createMemo(() => { if (session()?.parentID) return [] return data.session.family(route.sessionID).filter((id) => id !== route.sessionID) @@ -915,8 +925,8 @@ export function Session() { /> )} - - {(text) => } + + {(compaction) => } @@ -1084,7 +1094,7 @@ function SessionMessageView(props: { message: SessionMessage }) { - + } /> ) @@ -1265,14 +1275,51 @@ function SessionSkillMessage(props: { message: Extract + status?: "running" + text?: string +}) { + const ctx = use() + const { theme, syntax } = useTheme() + const status = () => props.message?.status ?? props.status + const text = () => props.message?.summary ?? props.text ?? "" return ( - - - {props.text} - - + + + ◇ Compaction queued + + + + + + + + Compacting conversation... + + + + + + + + + ) } diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 5b7652d22f8a..2fe978b245c2 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -77,7 +77,9 @@ export function createSessionRows(sessionID: Accessor) { .list(sessionID()) .flatMap((message) => message.type === "user" - ? [{ id: message.id, created: message.time.created, queued: message.metadata?.queued === true }] + ? [{ id: message.id, created: message.time.created, state: message.metadata?.queued ? "queued" : "visible" }] + : message.type === "compaction" + ? [{ id: message.id, created: message.time.created, state: message.status }] : [], ), () => setRows(reconcile(reduce())), @@ -88,9 +90,10 @@ export function createSessionRows(sessionID: Accessor) { setRows( produce((draft) => { if (draft.some((row) => row.type === "message" && row.messageID === messageID)) return - const queued = isQueued(messageID) - const index = queued ? draft.length : queuedStart(draft) - if (!queued) completePrevious(draft, index) + const pending = isPending(messageID) + const message = data.session.message.get(sessionID(), messageID) + const index = message?.type === "compaction" && pending ? queuedStart(draft) : pending ? draft.length : queuedStart(draft) + if (!pending) completePrevious(draft, index) draft.splice(index, 0, { type: "message", messageID }) }), ) @@ -145,13 +148,14 @@ export function createSessionRows(sessionID: Accessor) { return { messageID, partID: `${kind}:${Math.max(0, ordinal)}` } } - const isQueued = (messageID: string) => { + const isPending = (messageID: string) => { const message = data.session.message.get(sessionID(), messageID) - return message?.type === "user" && message.metadata?.queued === true + if (message?.type === "user") return message.metadata?.queued === true + return message?.type === "compaction" && (message.status === "queued" || message.status === "running") } const queuedStart = (rows: SessionRow[]) => { - const index = rows.findIndex((row) => row.type === "message" && isQueued(row.messageID)) + const index = rows.findIndex((row) => row.type === "message" && isPending(row.messageID)) return index === -1 ? rows.length : index } @@ -163,6 +167,7 @@ export function createSessionRows(sessionID: Accessor) { } const subscriptions = [ data.on("session.prompt.admitted", input), + data.on("session.compaction.admitted", input), data.on("session.context.updated", message), data.on("session.synthetic", (event) => { if (event.data.sessionID === sessionID() && event.data.description?.trim()) @@ -211,9 +216,13 @@ export function createSessionRows(sessionID: Accessor) { } export function reduceSessionRows(messages: SessionMessage[]) { - return [...messages.filter((message) => !isQueuedMessage(message)), ...messages.filter(isQueuedMessage)].reduce< - SessionRow[] - >((rows, message) => { + const pendingCompactions = messages.filter( + (message) => message.type === "compaction" && (message.status === "queued" || message.status === "running"), + ) + const queuedUsers = messages.filter(isQueuedMessage) + const pending = new Set([...pendingCompactions, ...queuedUsers].map((message) => message.id)) + return [...messages.filter((message) => !pending.has(message.id)), ...pendingCompactions, ...queuedUsers].reduce( + (rows, message) => { if (message.type !== "assistant") { if (message.type === "synthetic" && !message.description?.trim()) return rows if (!isQueuedMessage(message)) completePrevious(rows) @@ -231,7 +240,9 @@ export function reduceSessionRows(messages: SessionMessage[]) { rows.push({ type: "assistant-footer", messageID: message.id }) } return rows - }, []) + }, + [], + ) } export function resolvePart(message: SessionMessageAssistant, partID: string) { diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index b35490809142..53223dbb6e10 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -557,6 +557,46 @@ test("tracks session status from active sessions and execution events", async () await wait(() => data.session.status("session-retry") === "idle") expect(data.session.message.get("session-retry", "message-retry")).not.toHaveProperty("retry") + emitEvent(events, { + id: "evt_compaction_admitted", + created: 0, + type: "session.compaction.admitted", + durable: durable("session-manual", 1), + data: { sessionID: "session-manual", inputID: "message-compaction" }, + }) + await wait(() => { + const message = data.session.message.get("session-manual", "message-compaction") + return message?.type === "compaction" && message.status === "queued" + }) + emitEvent(events, { + id: "evt_manual_compaction_started", + created: 1, + type: "session.compaction.started", + durable: durable("session-manual", 2), + data: { sessionID: "session-manual", reason: "manual" }, + }) + emitEvent(events, { + id: "evt_manual_compaction_delta", + created: 2, + type: "session.compaction.delta", + data: { sessionID: "session-manual", text: "Streamed summary" }, + }) + await wait(() => { + const message = data.session.message.get("session-manual", "message-compaction") + return message?.type === "compaction" && message.summary === "Streamed summary" + }) + emitEvent(events, { + id: "evt_manual_compaction_ended", + created: 3, + type: "session.compaction.ended", + durable: durable("session-manual", 3), + data: { sessionID: "session-manual", reason: "manual", text: "Streamed summary", recent: "recent" }, + }) + await wait(() => { + const message = data.session.message.get("session-manual", "message-compaction") + return message?.type === "compaction" && message.status === "completed" + }) + emitEvent(events, { id: "evt_compaction_started", created: 0, diff --git a/packages/tui/test/cli/tui/session-rows.test.ts b/packages/tui/test/cli/tui/session-rows.test.ts index e55e42231b76..a7140579b036 100644 --- a/packages/tui/test/cli/tui/session-rows.test.ts +++ b/packages/tui/test/cli/tui/session-rows.test.ts @@ -209,6 +209,35 @@ test("renders a footer for a pre-output retry assistant after replay", () => { expect(reduceSessionRows([message])).toEqual([{ type: "assistant-footer", messageID: "assistant-retry" }]) }) +test("places a pending compaction barrier before every queued user message", () => { + const queued = (id: string, text: string, created: number): SessionMessage => ({ + type: "user", + id, + text, + metadata: { queued: true }, + time: { created }, + }) + const messages: SessionMessage[] = [ + queued("user-before", "Before", 1), + { + type: "compaction", + id: "compaction", + status: "queued", + reason: "manual", + summary: "", + recent: "", + time: { created: 2 }, + }, + queued("user-after", "After", 3), + ] + + expect(reduceSessionRows(messages)).toEqual([ + { type: "message", messageID: "compaction" }, + { type: "message", messageID: "user-before" }, + { type: "message", messageID: "user-after" }, + ]) +}) + function assistant(id: string, content: SessionMessageAssistant["content"]): SessionMessageAssistant { return { type: "assistant", diff --git a/specs/v2/session.md b/specs/v2/session.md index b9106bb4dc87..ce081c271d2d 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -32,7 +32,7 @@ sessions.active() -> absence means inactive; activity is not durable across process restarts ``` -`session_input` is the durable admission inbox. `PromptAdmitted` records and projects accepted input so pending queue state can be replayed, replicated, and observed by clients. Admitted inputs remain outside model-visible Session history until the serialized runner publishes `Prompted`. Its projector atomically writes the visible user message and marks the inbox row promoted in the same event transaction. The V1-to-V2 shadow bridge publishes the same `Prompted` event for already-visible V1 prompts. +`session_input` is the typed durable admission inbox for prompts and Session control operations. `PromptAdmitted` records accepted user input; `Compaction.Admitted` records one coalesced manual compaction barrier. Admitted prompts remain outside model-visible Session history until the serialized runner publishes `Prompted`. Its projector atomically writes the visible user message and marks the inbox row promoted in the same event transaction. A pending compaction blocks all unpromoted prompts, runs before the Session would otherwise become idle, and releases the backlog only after its durable ended or failed event settles the barrier. The V1-to-V2 shadow bridge publishes the same `Prompted` event for already-visible V1 prompts. `admittedSeq` is the durable Session event sequence of `PromptAdmitted`. Clients may use the admission event to represent queued input before `Prompted` makes it part of visible conversation history. @@ -110,7 +110,6 @@ Current Context Epoch follow-ups: - Add configured, remote, and nested instruction sources with explicit precedence and removal semantics. - Add durable post-crash continuation recovery for promoted or provider-dispatched work. -- Add explicit manual compaction on top of automatic request-budget compaction. - Add operational metrics for observation latency, unavailable sources, contention, baseline size, and chronological-update growth. - Consider watcher-backed per-file caching only if measurements show direct safe-boundary observation is too expensive. - Expose plugin-defined Context Sources only after plugin reload and scoped cleanup semantics are designed. @@ -124,7 +123,7 @@ Compaction keeps the full transcript durable while replacing its active model re The rolling summary is a continuation checkpoint with this complete heading order: `Continuation Goal`, `Operating Constraints`, `Progress` (`Completed`, `In Flight`, `Blocked`), `Decisions To Preserve`, `Resume From Here`, `Context To Preserve`, and `Working Files`. Every heading remains present even when its value is `(none)`. -`session.compaction.started.1` durably identifies the attempt. Compaction deltas are live-only progress. `session.compaction.ended.1` durably stores the final summary and serialized recent context; only this completed event projects a model-visible compaction message. On the next physical attempt, the runner observes that completed compaction and directly renders a fresh Context Epoch baseline. A failed or interrupted attempt therefore leaves the previous history boundary active. +`session.compaction.admitted.1` durably records a manual request and projects its queued transcript row. `session.compaction.started.1` identifies the attempt and transforms that row into a running divider. Compaction deltas are live-only progress rendered beneath it. `session.compaction.ended.1` durably stores the final summary and serialized recent context, completes the same row, and settles the manual barrier. `session.compaction.failed.1` settles an unsuccessful manual barrier without changing the previous history boundary. On the next physical attempt, the runner observes a completed compaction and directly renders a fresh Context Epoch baseline. Assistant text and reasoning follow a strict `started` / live-only `delta` / durable full-value `ended` lifecycle. A publisher permits at most one open fragment of each kind in a step and fails on a second start before the matching end. Provider block IDs remain internal to LLM adapters; each fragment event carries a Session-assigned kind-specific ordinal, matching the ordinal derived from projected content. UI identity is therefore the assistant message ID plus content kind and ordinal. Tool calls retain step-scoped `callID` because settlements and provider replay correlate through it. From beecb243bcd46cffdbc612b100430bd54e9a4ad4 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 4 Jul 2026 20:13:43 -0400 Subject: [PATCH 2/3] fix(core): harden compaction admission --- .../client/src/promise/generated/types.ts | 2 -- packages/client/test/effect.test.ts | 1 - packages/client/test/promise.test.ts | 1 - packages/core/src/session/input.ts | 24 ++++++++++++++----- packages/schema/src/session-input.ts | 9 +++++-- packages/sdk/js/src/v2/gen/types.gen.ts | 2 -- 6 files changed, 25 insertions(+), 14 deletions(-) diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 7804428e3575..7fff47d41ac3 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -600,7 +600,6 @@ export type SessionPromptInput = { export type SessionPromptOutput = { readonly data: { - readonly type: "prompt" readonly admittedSeq: number readonly id: string readonly sessionID: string @@ -801,7 +800,6 @@ export type SessionCommandInput = { export type SessionCommandOutput = { readonly data: { - readonly type: "prompt" readonly admittedSeq: number readonly id: string readonly sessionID: string diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index 40765d7be89e..17c5ca322188 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -212,7 +212,6 @@ const session = { const admission = { data: { - type: "prompt", admittedSeq: 0, id: "msg_test", sessionID: "ses_test", diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index ca2d5bba1b36..d28174c5365f 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -309,7 +309,6 @@ const session = { const admission = { data: { - type: "prompt", admittedSeq: 0, id: "msg_test", sessionID: "ses_test", diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index c5d6cb7cc3ab..65e3019b78fc 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/input.ts @@ -2,7 +2,7 @@ export * as SessionInput from "./input" import { and, asc, eq, isNull } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" -import { Admitted, Compaction, Delivery, Entry } from "@opencode-ai/schema/session-input" +import { Admitted, Compaction, Delivery, Entry, PromptEntry } from "@opencode-ai/schema/session-input" import type { Database } from "../database/database" import type { EventV2 } from "../event" import { KeyedMutex } from "../effect/keyed-mutex" @@ -14,7 +14,7 @@ import { SessionInputTable, SessionMessageTable } from "./sql" type DatabaseService = Database.Interface["db"] -export { Admitted, Compaction, Delivery, Entry } +export { Admitted, Compaction, Delivery, Entry, PromptEntry } const decodePrompt = Schema.decodeUnknownSync(Prompt) const encodePrompt = Schema.encodeSync(Prompt) @@ -38,7 +38,7 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Entry => { ...(row.promoted_seq === null ? {} : { handledSeq: row.promoted_seq }), }) if (!row.prompt || !row.delivery) throw new LifecycleConflict({ id: base.id }) - return Admitted.make({ + return PromptEntry.make({ ...base, type: "prompt", prompt: decodePrompt(row.prompt), @@ -47,6 +47,17 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Entry => { }) } +const toAdmitted = (entry: PromptEntry): Admitted => + Admitted.make({ + admittedSeq: entry.admittedSeq, + id: entry.id, + sessionID: entry.sessionID, + prompt: entry.prompt, + delivery: entry.delivery, + timeCreated: entry.timeCreated, + ...(entry.promotedSeq === undefined ? {} : { promotedSeq: entry.promotedSeq }), + }) + 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) return row === undefined ? undefined : fromRow(row) @@ -88,7 +99,7 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( const existing = yield* find(db, input.id) if (existing !== undefined) { if (existing.type !== "prompt") return yield* Effect.die(new LifecycleConflict({ id: input.id })) - return existing + return toAdmitted(existing) } return yield* events .publish(SessionEvent.PromptAdmitted, { @@ -103,7 +114,6 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( ? Effect.die(new Error("Prompt admission event is missing aggregate sequence")) : Effect.succeed( Admitted.make({ - type: "prompt", admittedSeq: event.durable.seq, id: input.id, sessionID: input.sessionID, @@ -115,7 +125,9 @@ export const admit = Effect.fn("SessionInput.admit")(function* ( ), Effect.catchDefect((defect) => find(db, input.id).pipe( - Effect.flatMap((stored) => (stored?.type === "prompt" ? Effect.succeed(stored) : Effect.die(defect))), + Effect.flatMap((stored) => + stored?.type === "prompt" ? Effect.succeed(toAdmitted(stored)) : Effect.die(defect), + ), ), ), ) diff --git a/packages/schema/src/session-input.ts b/packages/schema/src/session-input.ts index a3e691dcd165..a59c00b40e7d 100644 --- a/packages/schema/src/session-input.ts +++ b/packages/schema/src/session-input.ts @@ -13,7 +13,6 @@ export type Delivery = SessionDelivery.Delivery export interface Admitted extends Schema.Schema.Type {} export const Admitted = Schema.Struct({ - type: Schema.Literal("prompt"), admittedSeq: NonNegativeInt, id: SessionMessage.ID, sessionID: SessionID, @@ -23,6 +22,12 @@ export const Admitted = Schema.Struct({ promotedSeq: NonNegativeInt.pipe(optional), }).annotate({ identifier: "SessionInput.Admitted" }) +export interface PromptEntry extends Schema.Schema.Type {} +export const PromptEntry = Schema.Struct({ + type: Schema.Literal("prompt"), + ...Admitted.fields, +}).annotate({ identifier: "SessionInput.PromptEntry" }) + export interface Compaction extends Schema.Schema.Type {} export const Compaction = Schema.Struct({ type: Schema.Literal("compaction"), @@ -33,5 +38,5 @@ export const Compaction = Schema.Struct({ handledSeq: NonNegativeInt.pipe(optional), }).annotate({ identifier: "SessionInput.Compaction" }) -export const Entry = Schema.Union([Admitted, Compaction]).pipe(Schema.toTaggedUnion("type")) +export const Entry = Schema.Union([PromptEntry, Compaction]).pipe(Schema.toTaggedUnion("type")) export type Entry = typeof Entry.Type diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 2cfd67fe3906..9308e3028240 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -4399,7 +4399,6 @@ export type PromptInputFileAttachment = { } export type SessionInputAdmitted = { - type: "prompt" admittedSeq: number id: string sessionID: string @@ -8697,7 +8696,6 @@ export type SessionV2InfoV2 = { } export type SessionInputAdmittedV2 = { - type: "prompt" admittedSeq: number id: string sessionID: string From 41ffe60dab38ba23847a1c7d41845bbb184e4464 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 4 Jul 2026 21:55:33 -0400 Subject: [PATCH 3/3] fix(tui): align compaction separator states --- packages/tui/src/routes/session/index.tsx | 77 ++++++++++++----------- 1 file changed, 42 insertions(+), 35 deletions(-) diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 5411368e8e97..bd2a90ece57a 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -21,7 +21,7 @@ import { useProject } from "../../context/project" import { useData } from "../../context/data" import { SplitBorder } from "../../ui/border" import { useTuiPaths, useTuiTerminalEnvironment } from "../../context/runtime" -import { Spinner } from "../../component/spinner" +import { Spinner, SPINNER_FRAMES } from "../../component/spinner" import { createSyntaxStyleMemo, generateSubtleSyntax, selectedForeground, useTheme } from "../../context/theme" import { BoxRenderable, ScrollBoxRenderable, addDefaultParsers, TextAttributes, RGBA } from "@opentui/core" import { Prompt, type PromptRef } from "../../component/prompt" @@ -1281,45 +1281,52 @@ function CompactionMessage(props: { text?: string }) { const ctx = use() + const kv = useKV() const { theme, syntax } = useTheme() const status = () => props.message?.status ?? props.status const text = () => props.message?.summary ?? props.text ?? "" + const color = () => (status() === "failed" ? theme.error : status() === "completed" ? theme.success : theme.textMuted) + const border = () => (status() === "queued" ? theme.border : color()) return ( - - - ◇ Compaction queued - - - - - - - - Compacting conversation... - - - - - - + + + + + + + ⋯}> + + + + + ✓ + + + ✗ + + + ◇ + + + {status() === "queued" ? "Compaction queued" : "Compaction"} - - + + + + + + + + ) }