diff --git a/packages/app/src/context/server-sdk.test.ts b/packages/app/src/context/server-sdk.test.ts index 7b592178fa2..052faf496a5 100644 --- a/packages/app/src/context/server-sdk.test.ts +++ b/packages/app/src/context/server-sdk.test.ts @@ -126,8 +126,9 @@ describe("enqueueServerEvent", () => { enqueue(partUpdated("old")) enqueue({ + id: "event-delete", type: "session.deleted", - properties: { sessionID: "session", info: { id: "session" } }, + properties: { sessionID: "session" }, } as Event) enqueue(partUpdated("new")) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 2f8f00526fc..588d67cad6e 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -168,8 +168,11 @@ export type Endpoint4_13Output = EffectValue = (input: Endpoint4_13Input) => Effect.Effect type Endpoint4_14Request = Parameters[0] -export type Endpoint4_14Input = { readonly sessionID: Endpoint4_14Request["params"]["sessionID"] } -export type Endpoint4_14Output = EffectValue> +export type Endpoint4_14Input = { + readonly sessionID: Endpoint4_14Request["params"]["sessionID"] + readonly id?: Endpoint4_14Request["payload"]["id"] +} +export type Endpoint4_14Output = EffectValue>["data"] export type SessionCompactOperation = (input: Endpoint4_14Input) => Effect.Effect type Endpoint4_15Request = Parameters[0] diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index b138af8f65a..945bd0ffc09 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -229,9 +229,15 @@ const Endpoint4_13 = (raw: RawClient["server.session"]) => (input: Endpoint4_13I }).pipe(Effect.mapError(mapClientError)) type Endpoint4_14Request = Parameters[0] -type Endpoint4_14Input = { readonly sessionID: Endpoint4_14Request["params"]["sessionID"] } +type Endpoint4_14Input = { + readonly sessionID: Endpoint4_14Request["params"]["sessionID"] + readonly id?: Endpoint4_14Request["payload"]["id"] +} const Endpoint4_14 = (raw: RawClient["server.session"]) => (input: Endpoint4_14Input) => - 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_15Request = Parameters[0] type Endpoint4_15Input = { readonly sessionID: Endpoint4_15Request["params"]["sessionID"] } diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 7e18a0c061c..697d54d732d 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -556,16 +556,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 0d66ca1fc98..7482d237258 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 @@ -879,9 +879,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"] } @@ -1095,6 +1107,7 @@ export type SessionContextOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -1573,6 +1586,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 @@ -1596,6 +1618,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 @@ -1830,6 +1861,7 @@ export type SessionMessageOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -2034,6 +2066,7 @@ export type MessageListOutput = { } | { readonly type: "compaction" + readonly status: "queued" | "running" | "completed" | "failed" readonly reason: "auto" | "manual" readonly summary: string readonly recent: string @@ -4888,6 +4921,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 @@ -4919,6 +4961,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 4c5ea512d79..6247d256105 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -140,6 +140,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: [] }))) } @@ -148,10 +151,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => { } if (url.endsWith("/api/session/active")) { return Effect.succeed( - HttpClientResponse.fromWeb( - request, - Response.json({ data: { ses_test: { type: "running" } } }), - ), + HttpClientResponse.fromWeb(request, Response.json({ data: { ses_test: { type: "running" } } })), ) } if (request.method === "POST" && url.endsWith("/api/session")) { @@ -161,10 +161,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => { return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status: 204 }))) } return Effect.succeed( - HttpClientResponse.fromWeb( - request, - Response.json({ data: [session.data], cursor: { next: "next" } }), - ), + HttpClientResponse.fromWeb(request, Response.json({ data: [session.data], cursor: { next: "next" } })), ) }) const result = await Effect.gen(function* () { @@ -268,6 +265,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 21b191fd2ab..4f2419075f8 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -46,7 +46,7 @@ test("exposes every standard HTTP API group", () => { expect(Object.keys(client.file)).toEqual(["read", "list", "find"]) expect(Object.keys(client.vcs)).toEqual(["status", "diff"]) expect(Object.keys(client.pty)).toEqual(["list", "create", "get", "update", "remove"]) - expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "output", "remove"]) + expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "timeout", "output", "remove"]) expect(Object.keys(client.project)).toEqual(["list", "current", "directories"]) }) @@ -240,10 +240,10 @@ 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")) - return Response.json({ data: { ses_test: { type: "running" } } }) + if (url.endsWith("/api/session/active")) return Response.json({ data: { ses_test: { type: "running" } } }) if (init?.method === "POST" && url.endsWith("/api/session")) return Response.json(session) if (init?.method === "POST") return new Response(null, { status: 204 }) return Response.json({ data: [session.data], cursor: { next: "next" } }) @@ -364,6 +364,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 4bd411e53bf..d4199094442 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,10 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "95328a41-789d-44de-9643-6ac6ecd6b4ec", - "prevIds": [ - "992b24b9-f3e9-41f5-87a5-4917d1423169" - ], + "id": "b0355fd9-bf41-42e3-9dca-76107de27ecd", + "prevIds": ["95328a41-789d-44de-9643-6ac6ecd6b4ec"], "ddl": [ { "name": "workspace", @@ -1012,13 +1010,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, @@ -1567,13 +1575,9 @@ "table": "session_share" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1582,13 +1586,9 @@ "table": "workspace" }, { - "columns": [ - "active_account_id" - ], + "columns": ["active_account_id"], "tableTo": "account", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "SET NULL", "nameExplicit": false, @@ -1597,13 +1597,9 @@ "table": "account_state" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "tableTo": "event_sequence", - "columnsTo": [ - "aggregate_id" - ], + "columnsTo": ["aggregate_id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1612,13 +1608,9 @@ "table": "event" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1627,13 +1619,9 @@ "table": "permission" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1642,13 +1630,9 @@ "table": "project_directory" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1657,13 +1641,9 @@ "table": "instruction_checkpoint" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1672,13 +1652,9 @@ "table": "instruction_entry" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1687,13 +1663,9 @@ "table": "message" }, { - "columns": [ - "message_id" - ], + "columns": ["message_id"], "tableTo": "message", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1702,13 +1674,9 @@ "table": "part" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1717,13 +1685,9 @@ "table": "session_input" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1732,13 +1696,9 @@ "table": "session_message" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1747,13 +1707,9 @@ "table": "session" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1762,13 +1718,9 @@ "table": "todo" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1777,184 +1729,140 @@ "table": "session_share" }, { - "columns": [ - "email", - "url" - ], + "columns": ["email", "url"], "nameExplicit": false, "name": "control_account_pk", "entityType": "pks", "table": "control_account" }, { - "columns": [ - "project_id", - "directory" - ], + "columns": ["project_id", "directory"], "nameExplicit": false, "name": "project_directory_pk", "entityType": "pks", "table": "project_directory" }, { - "columns": [ - "session_id", - "key" - ], + "columns": ["session_id", "key"], "nameExplicit": false, "name": "instruction_entry_pk", "entityType": "pks", "table": "instruction_entry" }, { - "columns": [ - "session_id", - "position" - ], + "columns": ["session_id", "position"], "nameExplicit": false, "name": "todo_pk", "entityType": "pks", "table": "todo" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "workspace_pk", "table": "workspace", "entityType": "pks" }, { - "columns": [ - "name" - ], + "columns": ["name"], "nameExplicit": false, "name": "data_migration_pk", "table": "data_migration", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_state_pk", "table": "account_state", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_pk", "table": "account", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "credential_pk", "table": "credential", "entityType": "pks" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "nameExplicit": false, "name": "event_sequence_pk", "table": "event_sequence", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "event_pk", "table": "event", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "permission_pk", "table": "permission", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "project_pk", "table": "project", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "instruction_checkpoint_pk", "table": "instruction_checkpoint", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "message_pk", "table": "message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "part_pk", "table": "part", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_input_pk", "table": "session_input", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_message_pk", "table": "session_message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_pk", "table": "session", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "session_share_pk", "table": "session_share", @@ -2086,6 +1994,10 @@ "value": "promoted_seq", "isExpression": false }, + { + "value": "type", + "isExpression": false + }, { "value": "delivery", "isExpression": false @@ -2098,7 +2010,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" }, @@ -2272,4 +2198,4 @@ } ], "renames": [] -} \ No newline at end of file +} diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index e8418683359..99ca206c1f4 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -47,5 +47,6 @@ export const migrations = ( import("./migration/20260703200000_reset_v2_session_events"), import("./migration/20260705180000_rename_instructions"), import("./migration/20260706223930_add-session-fork"), + import("./migration/20260707010146_durable_session_inbox"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260707010146_durable_session_inbox.ts b/packages/core/src/database/migration/20260707010146_durable_session_inbox.ts new file mode 100644 index 00000000000..e4909926208 --- /dev/null +++ b/packages/core/src/database/migration/20260707010146_durable_session_inbox.ts @@ -0,0 +1,43 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260707010146_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 24092623164..0bb6050bf6e 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, @@ -261,7 +262,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 827adb261fb..b5d28c46127 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" @@ -96,6 +95,7 @@ type CreateInput = CreateBaseInput & ({ location: Location.Ref; parentID?: never } | { parentID: SessionSchema.ID; location?: never }) type CompactInput = { + id?: SessionMessage.ID sessionID: SessionSchema.ID } @@ -125,6 +125,13 @@ export class AttachmentError extends Schema.TaggedErrorClass()( uri: Schema.String, message: Schema.String, }) {} +export class CompactionConflictError extends Schema.TaggedErrorClass()( + "Session.CompactionConflictError", + { + sessionID: SessionSchema.ID, + inputID: SessionMessage.ID, + }, +) {} export class BusyError extends Schema.TaggedErrorClass()("Session.BusyError", { sessionID: SessionSchema.ID, }) {} @@ -140,6 +147,7 @@ export type Error = | OperationUnavailableError | PromptConflictError | AttachmentError + | CompactionConflictError | BusyError | SkillNotFoundError | CommandV2.NotFoundError @@ -224,7 +232,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 @@ -626,19 +634,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) @@ -734,11 +743,7 @@ function synthesizeTerminalShellInfo(started: ShellSchema.Info): ShellSchema.Inf const resolvePrompt = Effect.fn("V2Session.resolvePrompt")(function* (input: PromptInput.Prompt) { const fs = yield* FSUtil.Service const files = input.files - ? yield* Effect.forEach( - input.files, - (file) => materializeAttachment(fs, file), - { concurrency: 8 }, - ) + ? yield* Effect.forEach(input.files, (file) => materializeAttachment(fs, file), { concurrency: 8 }) : undefined return Prompt.make({ text: input.text, agents: input.agents, files }) }) @@ -769,7 +774,11 @@ const materializeAttachment = Effect.fn("V2Session.materializeAttachment")(funct const content = mime === "text/plain" && resolved.start !== undefined ? Buffer.from( - Buffer.from(resolved.bytes).toString("utf8").split("\n").slice(resolved.start - 1, resolved.end).join("\n"), + Buffer.from(resolved.bytes) + .toString("utf8") + .split("\n") + .slice(resolved.start - 1, resolved.end) + .join("\n"), ) : resolved.bytes return FileAttachment.create({ @@ -799,13 +808,13 @@ const readFileAttachment = Effect.fn("V2Session.readFileAttachment")(function* ( }, catch: () => new AttachmentError({ uri, message: `Invalid file URI: ${uri}` }), }) - const info = yield* fs.stat(target).pipe( - Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })), - ) + const info = yield* fs + .stat(target) + .pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` }))) if (info.type === "Directory") { - const entries = yield* fs.readDirectoryEntries(target).pipe( - Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })), - ) + const entries = yield* fs + .readDirectoryEntries(target) + .pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` }))) return { bytes: Buffer.from( entries @@ -827,9 +836,9 @@ const readFileAttachment = Effect.fn("V2Session.readFileAttachment")(function* ( uri, message: `Attachment exceeds the ${MAX_ATTACHMENT_BYTES} byte limit: ${uri}`, }) - const bytes = yield* fs.readFile(target).pipe( - Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })), - ) + const bytes = yield* fs + .readFile(target) + .pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` }))) return { bytes, source: { type: "uri" as const, uri }, start, end, name: path.basename(target), mime: undefined } }) diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 63f26a18620..e90b6455dce 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -260,7 +260,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 b3f0b6c741d..e719366940d 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 72622d32ef7..c08439d0c17 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, PromptEntry } 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 "@opencode-ai/schema/prompt" @@ -13,30 +14,77 @@ import { SessionInputTable, SessionMessageTable } from "./sql" type DatabaseService = Database.Interface["db"] -export { Admitted, Delivery } +export { Admitted, Compaction, Delivery, Entry, PromptEntry } 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 PromptEntry.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 }), }) +} + +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) }) -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 +97,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 toAdmitted(existing) + } return yield* events .publish(SessionEvent.PromptAdmitted, { inputID: input.id, @@ -73,11 +124,54 @@ 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(toAdmitted(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 +195,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 +208,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 +254,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 +262,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 +271,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 +345,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 +390,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 +413,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 5f2e69b9e72..96881e7c699 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) @@ -424,21 +443,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 d813e135448..625b6f1553f 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -248,6 +248,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)) @@ -297,11 +298,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, @@ -431,8 +433,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) @@ -642,6 +666,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)) @@ -672,7 +710,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 46c5eb7297a..2e05d1edcf7 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -228,7 +228,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, { @@ -511,11 +514,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 @@ -539,11 +567,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 bce9443d121..f4e1d7ca01c 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -209,6 +209,7 @@ function toLLMMessage(message: SessionMessage.Message, model: ModelV2.Ref): Mess 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 97b908bee18..e7d3bdca58b 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" @@ -147,8 +148,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() @@ -156,12 +158,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 c5f783e5749..4f147c0f3d1 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/20260707010146_durable_session_inbox" import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions" import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" @@ -109,13 +110,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" }, @@ -353,7 +355,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, '{}')`, @@ -416,6 +418,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 df63072fb9a..40d9f48a199 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -75,7 +75,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 @@ -95,13 +95,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 31909c0e765..3e76ddbb424 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -246,7 +246,9 @@ describe("SessionV2.prompt", () => { mention: { start: 8, end: 17, text: "[Image 1]" }, }, ]) - 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) }), ) @@ -336,7 +338,8 @@ describe("SessionV2.prompt", () => { name: "image.png", }, ]) - expect((yield* admitted(message.id))?.prompt.files).toEqual(message.prompt.files) + const stored = yield* admitted(message.id) + expect(stored?.type === "prompt" ? stored.prompt.files : undefined).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 a7852c292a1..b34d0e9918a 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -106,6 +106,7 @@ describe("toLLMMessages", () => { SessionMessage.Compaction.make({ id: id("compaction"), type: "compaction", + status: "completed", reason: "auto", summary: "Earlier work", recent: "Recent work", diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 969d4fabbe7..ec893a55272 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1324,6 +1324,101 @@ 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: PromptInput.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: PromptInput.Prompt.make({ text: "Steer after compaction" }), + resume: false, + }) + yield* session.prompt({ + sessionID, + prompt: PromptInput.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: PromptInput.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: PromptInput.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 74e2c1c75c6..090df62eb69 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -305,13 +305,7 @@ export const makeSessionGroup = (sessionLo resume: Schema.Boolean.pipe(Schema.optional), }), success: Schema.Struct({ data: SessionInput.Admitted }), - error: [ - ConflictError, - InvalidRequestError, - SessionNotFoundError, - CommandNotFoundError, - CommandEvaluationError, - ], + error: [ConflictError, InvalidRequestError, SessionNotFoundError, CommandNotFoundError, CommandEvaluationError], }) .middleware(sessionLocationMiddleware) .annotateMerge( @@ -386,15 +380,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 64f691e7ceb..8e96321ce2a 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -436,6 +436,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, @@ -466,6 +476,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 { @@ -517,9 +534,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 eefe68be2b6..a59c00b40e7 100644 --- a/packages/schema/src/session-input.ts +++ b/packages/schema/src/session-input.ts @@ -21,3 +21,22 @@ export const Admitted = Schema.Struct({ timeCreated: DateTimeUtcFromMillis, 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"), + admittedSeq: NonNegativeInt, + id: SessionMessage.ID, + sessionID: SessionID, + timeCreated: DateTimeUtcFromMillis, + handledSeq: NonNegativeInt.pipe(optional), +}).annotate({ identifier: "SessionInput.Compaction" }) + +export const Entry = Schema.Union([PromptEntry, 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 aafef1dba2a..62e1a1038d3 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 a62c0c78586..1a5cbe7fa73 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -131,8 +131,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 d2e22a3fcb1..458f9c2d2ea 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -448,6 +448,8 @@ import type { V2ShellOutputResponses, V2ShellRemoveErrors, V2ShellRemoveResponses, + V2ShellTimeoutErrors, + V2ShellTimeoutResponses, V2SkillListErrors, V2SkillListResponses, V2VcsDiffErrors, @@ -6305,19 +6307,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, + }, }) } @@ -7787,6 +7805,46 @@ export class Shell extends HeyApiClient { }) } + /** + * Update shell timeout + * + * Replace a running shell command's timeout from now, or clear it with zero. + */ + public timeout( + parameters: { + id: string + location?: { + directory?: string | null + workspace?: string | null + } | null + timeout?: number + }, + options?: Options, + ) { + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "path", key: "id" }, + { in: "query", key: "location" }, + { in: "body", key: "timeout" }, + ], + }, + ], + ) + return (options?.client ?? this.client).patch({ + url: "/api/shell/{id}/timeout", + ...options, + ...params, + headers: { + "Content-Type": "application/json", + ...options?.headers, + ...params.headers, + }, + }) + } + /** * Read shell output * diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index bdc0f9dc95a..87aa5ddbe1c 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" @@ -1772,8 +1789,10 @@ export type GlobalEvent = { | SyncEventSessionToolSuccess | SyncEventSessionToolFailed | SyncEventSessionRetryScheduled + | SyncEventSessionCompactionAdmitted | SyncEventSessionCompactionStarted | SyncEventSessionCompactionEnded + | SyncEventSessionCompactionFailed | SyncEventSessionRevertStaged | SyncEventSessionRevertCleared | SyncEventSessionRevertCommitted @@ -2940,8 +2959,10 @@ export type SessionDurableEvent = | SessionToolSuccess | SessionToolFailed | SessionRetryScheduled + | SessionCompactionAdmitted | SessionCompactionStarted | SessionCompactionEnded + | SessionCompactionFailed | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted @@ -3087,9 +3108,11 @@ export type V2Event = | SessionToolSuccess | SessionToolFailed | SessionRetryScheduled + | SessionCompactionAdmitted | SessionCompactionStarted | SessionCompactionDelta | SessionCompactionEnded + | SessionCompactionFailed | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted @@ -4168,6 +4191,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 @@ -4200,6 +4238,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 @@ -4374,6 +4426,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?: { @@ -4590,6 +4651,7 @@ export type SessionMessageAssistant = { export type SessionMessageCompaction = { type: "compaction" + status: "queued" | "running" | "completed" | "failed" reason: "auto" | "manual" summary: string recent: string @@ -5278,6 +5340,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 @@ -5318,6 +5399,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 @@ -7247,6 +7346,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" @@ -7276,6 +7384,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" @@ -8439,9 +8555,11 @@ export type V2EventV2 = | SessionToolSuccessV2 | SessionToolFailedV2 | SessionRetryScheduledV2 + | SessionCompactionAdmittedV2 | SessionCompactionStartedV2 | SessionCompactionDeltaV2 | SessionCompactionEndedV2 + | SessionCompactionFailedV2 | SessionRevertStagedV2 | SessionRevertClearedV2 | SessionRevertCommittedV2 @@ -8583,6 +8701,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?: { @@ -8721,6 +8848,7 @@ export type SessionMessageAssistantV2 = { export type SessionMessageCompactionV2 = { type: "compaction" + status: "queued" | "running" | "completed" | "failed" reason: "auto" | "manual" summary: string recent: string @@ -9416,6 +9544,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 @@ -9456,6 +9603,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 @@ -15426,7 +15591,9 @@ export type V2SessionShellResponses = { export type V2SessionShellResponse = V2SessionShellResponses[keyof V2SessionShellResponses] export type V2SessionCompactData = { - body?: never + body: { + id?: string | null + } path: { sessionID: string } @@ -15448,26 +15615,20 @@ export type V2SessionCompactErrors = { */ 404: SessionNotFoundError /** - * SessionBusyError + * ConflictError */ - 409: SessionBusyError - /** - * UnknownError - */ - 500: UnknownErrorV2 - /** - * ServiceUnavailableError - */ - 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] @@ -17796,7 +17957,7 @@ export type V2ShellCreateData = { body: { command: string cwd?: string - timeout?: number + timeout: number metadata?: { [key: string]: unknown } @@ -17919,6 +18080,51 @@ export type V2ShellGetResponses = { export type V2ShellGetResponse = V2ShellGetResponses[keyof V2ShellGetResponses] +export type V2ShellTimeoutData = { + body: { + timeout: number + } + path: { + id: string + } + query?: { + location?: { + directory?: string | null + workspace?: string | null + } | null + } + url: "/api/shell/{id}/timeout" +} + +export type V2ShellTimeoutErrors = { + /** + * InvalidRequestError + */ + 400: InvalidRequestErrorV2 + /** + * UnauthorizedError + */ + 401: UnauthorizedError + /** + * ShellNotFoundError + */ + 404: ShellNotFoundError +} + +export type V2ShellTimeoutError = V2ShellTimeoutErrors[keyof V2ShellTimeoutErrors] + +export type V2ShellTimeoutResponses = { + /** + * Success + */ + 200: { + location: LocationInfoV2 + data: ShellV2 + } +} + +export type V2ShellTimeoutResponse = V2ShellTimeoutResponses[keyof V2ShellTimeoutResponses] + export type V2ShellOutputData = { body?: never path: { diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 75227bb0070..ce3bc15f364 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -364,44 +364,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}`, - }), - ), - ), - Effect.catchTag("Session.OperationUnavailableError", (error) => - Effect.fail( - new ServiceUnavailableError({ - message: `Session ${error.operation} is not available yet`, - service: `session.${error.operation}`, - }), - ), - ), - 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 { + 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}`, + }), ), - ) - }), - ) - return HttpApiSchema.NoContent.make() + ), + Effect.catchTag("Session.CompactionConflictError", (error) => + Effect.fail( + new ConflictError({ + message: `Compaction input ID conflicts with an existing durable record: ${error.inputID}`, + resource: error.inputID, + }), + ), + ), + ), + } }), ) .handle( diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 581ebe17560..723f80633dd 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -55,6 +55,7 @@ type Data = { family: Record status: Record compaction: Partial> + compactionReason: Partial> message: Record input: Record permission: Record @@ -92,6 +93,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ family: {}, status: {}, compaction: {}, + compactionReason: {}, message: {}, input: {}, permission: {}, @@ -145,6 +147,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 => @@ -580,8 +588,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": @@ -589,6 +617,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 @@ -619,13 +649,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, @@ -633,6 +678,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, [ @@ -785,6 +838,14 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ].toSorted((a, b) => a.time.created - b.time.created) messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) setStore("session", "message", sessionID, messages) + const running = messages.find((message) => message.type === "compaction" && message.status === "running") + setStore("session", "compaction", sessionID, running?.type === "compaction" ? running.summary : undefined) + setStore( + "session", + "compactionReason", + sessionID, + running?.type === "compaction" ? running.reason : undefined, + ) }, }, permission: { diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index b9cf3ad81c2..6c2bfd34ef5 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, useTheme } from "../../context/theme" import { BoxRenderable, ScrollBoxRenderable, addDefaultParsers, TextAttributes, RGBA } from "@opentui/core" import { Prompt, type PromptRef } from "../../component/prompt" @@ -172,6 +172,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) @@ -926,8 +936,8 @@ export function Session() { /> )} - - {(text) => } + + {(compaction) => } @@ -1100,7 +1110,7 @@ function SessionMessageView(props: { message: SessionMessage }) { - + } /> ) @@ -1285,12 +1295,56 @@ function SessionSkillMessage(props: { message: Extract + status?: "running" + 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 ( - - - {props.text} + + + + + + + ⋯}> + + + + + ✓ + + + ✗ + + + ◇ + + + {status() === "queued" ? "Compaction queued" : "Compaction"} + + + + + + + ) diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 0bbcedd6fba..5d2a3614520 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -83,7 +83,15 @@ export function createSessionRows(sessionID: Accessor) { input: data.session.input.has(sessionID(), message.id), }, ] - : [], + : message.type === "compaction" + ? [ + { + id: message.id, + created: message.time.created, + input: message.status === "queued" || message.status === "running", + }, + ] + : [], ), () => setRows(reconcile(reduce())), ), @@ -93,9 +101,11 @@ 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 }) }), ) @@ -144,12 +154,14 @@ export function createSessionRows(sessionID: Accessor) { }), ) - const isQueued = (messageID: string) => { - return data.session.input.has(sessionID(), messageID) + const isPending = (messageID: string) => { + const message = data.session.message.get(sessionID(), messageID) + if (message?.type === "user") return data.session.input.has(sessionID(), messageID) + 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 } @@ -161,6 +173,7 @@ export function createSessionRows(sessionID: Accessor) { } const subscriptions = [ data.on("session.prompt.admitted", input), + data.on("session.compaction.admitted", input), data.on("session.instructions.updated", message), data.on("session.synthetic", (event) => { if (event.data.sessionID === sessionID() && event.data.description?.trim()) @@ -169,7 +182,9 @@ export function createSessionRows(sessionID: Accessor) { data.on("session.shell.started", message), data.on("session.agent.selected", message), data.on("session.model.selected", message), - data.on("session.compaction.ended", message), + data.on("session.compaction.ended", (event) => { + if (event.data.reason !== "manual") message(event) + }), data.on("session.text.delta", (event) => { if (event.data.sessionID === sessionID()) appendPart({ messageID: event.data.assistantMessageID, partID: `text:${event.data.ordinal}` }) @@ -211,28 +226,33 @@ export function createSessionRows(sessionID: Accessor) { export function reduceSessionRows(messages: SessionMessage[], inputs = new Set()) { const isInput = (message: SessionMessage) => inputs.has(message.id) - return [...messages.filter((message) => !isInput(message)), ...messages.filter(isInput)].reduce( - (rows, message) => { - if (message.type !== "assistant") { - if (message.type === "synthetic" && !message.description?.trim()) return rows - if (!isInput(message)) completePrevious(rows) - rows.push({ type: "message", messageID: message.id }) - return rows - } - const ordinals = { text: 0, reasoning: 0 } - message.content.forEach((part) => { - const partID = part.type === "tool" ? part.id : `${part.type}:${ordinals[part.type]++}` - if ((part.type === "text" || part.type === "reasoning") && !part.text.trim()) return - append(rows, { messageID: message.id, partID }, part) - }) - if ((message.finish && !["tool-calls", "unknown"].includes(message.finish)) || message.error || message.retry) { - completePrevious(rows) - rows.push({ type: "assistant-footer", messageID: message.id }) - } - return rows - }, - [], + const pendingCompactions = messages.filter( + (message) => message.type === "compaction" && (message.status === "queued" || message.status === "running"), ) + const pending = new Set([...pendingCompactions.map((message) => message.id), ...inputs]) + return [ + ...messages.filter((message) => !pending.has(message.id)), + ...pendingCompactions, + ...messages.filter(isInput), + ].reduce((rows, message) => { + if (message.type !== "assistant") { + if (message.type === "synthetic" && !message.description?.trim()) return rows + if (!pending.has(message.id)) completePrevious(rows) + rows.push({ type: "message", messageID: message.id }) + return rows + } + const ordinals = { text: 0, reasoning: 0 } + message.content.forEach((part) => { + const partID = part.type === "tool" ? part.id : `${part.type}:${ordinals[part.type]++}` + if ((part.type === "text" || part.type === "reasoning") && !part.text.trim()) return + append(rows, { messageID: message.id, partID }, part) + }) + if ((message.finish && !["tool-calls", "unknown"].includes(message.finish)) || message.error || message.retry) { + completePrevious(rows) + 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 8a8b6956ae4..b7b349a196d 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -108,6 +108,64 @@ test("refreshes resources into reactive getters", async () => { } }) +test("restores running manual compaction before applying live deltas", async () => { + const events = createEventStream() + const calls = createFetch((url) => { + if (url.pathname === "/api/session/session-compaction/message") + return json({ + data: [ + { + id: "message-compaction", + type: "compaction", + status: "running", + reason: "manual", + summary: "Existing ", + recent: "", + time: { created: 1 }, + }, + ], + cursor: {}, + }) + }, events) + let data!: ReturnType + + function Probe() { + data = useData() + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await data.session.message.refresh("session-compaction") + expect(data.session.compaction("session-compaction")).toBe("Existing ") + + emitEvent(events, { + id: "evt_compaction_delta", + created: 2, + type: "session.compaction.delta", + data: { sessionID: "session-compaction", text: "summary" }, + }) + + await wait(() => { + const message = data.session.message.get("session-compaction", "message-compaction") + return message?.type === "compaction" && message.summary === "Existing summary" + }) + } finally { + app.renderer.destroy() + } +}) + test("reconnects the event stream and bootstraps fresh data", async () => { const events = createEventStream() const requests = { active: 0, event: 0, model: 0 } @@ -402,10 +460,12 @@ test("tracks session status from active sessions and execution events", async () }, events) let data!: ReturnType let rows!: SessionRow[] + let manualRows!: SessionRow[] function Probe() { data = useData() rows = createSessionRows(() => "session-retry") + manualRows = createSessionRows(() => "session-manual") return } @@ -610,6 +670,49 @@ 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" + }) + expect(manualRows.filter((row) => row.type === "message")).toEqual([ + { type: "message", messageID: "message-compaction" }, + ]) + 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 e55e42231b7..362ab40369a 100644 --- a/packages/tui/test/cli/tui/session-rows.test.ts +++ b/packages/tui/test/cli/tui/session-rows.test.ts @@ -209,6 +209,34 @@ 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, + 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, new Set(["user-before", "user-after"]))).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 8057a7a8cdc..511ece8cfcd 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. @@ -120,7 +120,6 @@ Current instruction follow-ups: - Add configured and remote 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 step-boundary observation is too expensive. - Design any plugin-defined instruction contribution as an explicit runner composition boundary; do not reintroduce a registry implicitly. @@ -134,7 +133,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: `Objective`, `Important Details`, `Work State`, and `Next Move`. `Work State` records completed, active, and blocked work, while `Next Move` records the immediate and following actions. 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 instruction baseline through `InstructionCheckpoint`. 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 instruction baseline through `InstructionCheckpoint`. 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.