ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(core): session.pending.list API with pending-only session_pending storage (#36126)

Browse source
This commit is contained in:
Kit Langton 2026-07-09 17:25:40 -04:00 • committed by GitHub
commit a72992e00f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
43 changed files with 986 additions and 649 deletions
Download patch file Download diff file Expand all files Collapse all files

306
packages/core/schema.json
View file

@ -1,8 +1,8 @@
{
"version": "7",
"dialect": "sqlite",
"id": "01451b27-1e51-4657-b2d0-4b457dffa3ec",
"prevIds": ["8c1748cc-f978-4df4-879d-fe7fda5c4c34"],
"id": "2c759c08-79c8-4179-9e15-d2fea45c9dec",
"prevIds": ["01451b27-1e51-4657-b2d0-4b457dffa3ec"],
"ddl": [
{
"name": "workspace",
@ -65,11 +65,11 @@
"entityType": "tables"
},
{
"name": "session_input",
"name": "session_message",
"entityType": "tables"
},
{
"name": "session_message",
"name": "session_pending",
"entityType": "tables"
},
{
@ -980,86 +980,6 @@
"entityType": "columns",
"table": "part"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "id",
"entityType": "columns",
"table": "session_input"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "session_id",
"entityType": "columns",
"table": "session_input"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "type",
"entityType": "columns",
"table": "session_input"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "data",
"entityType": "columns",
"table": "session_input"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "delivery",
"entityType": "columns",
"table": "session_input"
},
{
"type": "integer",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "admitted_seq",
"entityType": "columns",
"table": "session_input"
},
{
"type": "integer",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "promoted_seq",
"entityType": "columns",
"table": "session_input"
},
{
"type": "integer",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "time_created",
"entityType": "columns",
"table": "session_input"
},
{
"type": "text",
"notNull": false,
@ -1130,6 +1050,76 @@
"entityType": "columns",
"table": "session_message"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "id",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "session_id",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "type",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "data",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "delivery",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "integer",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "admitted_seq",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "integer",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "time_created",
"entityType": "columns",
"table": "session_pending"
},
{
"type": "text",
"notNull": false,
@ -1616,9 +1606,9 @@
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
"name": "fk_session_input_session_id_session_id_fk",
"name": "fk_session_message_session_id_session_id_fk",
"entityType": "fks",
"table": "session_input"
"table": "session_message"
},
{
"columns": ["session_id"],
@ -1627,9 +1617,9 @@
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
"name": "fk_session_message_session_id_session_id_fk",
"name": "fk_session_input_session_id_session_id_fk",
"entityType": "fks",
"table": "session_message"
"table": "session_pending"
},
{
"columns": ["project_id"],
@ -1761,15 +1751,15 @@
{
"columns": ["id"],
"nameExplicit": false,
"name": "session_input_pk",
"table": "session_input",
"name": "session_message_pk",
"table": "session_message",
"entityType": "pks"
},
{
"columns": ["id"],
"nameExplicit": false,
"name": "session_message_pk",
"table": "session_message",
"name": "session_input_pk",
"table": "session_pending",
"entityType": "pks"
},
{
@ -1902,82 +1892,6 @@
"entityType": "indexes",
"table": "part"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "promoted_seq",
"isExpression": false
},
{
"value": "delivery",
"isExpression": false
},
{
"value": "admitted_seq",
"isExpression": false
}
],
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_input_session_pending_delivery_seq_idx",
"entityType": "indexes",
"table": "session_input"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
}
],
"isUnique": true,
"where": "\"session_input\".\"type\" = 'compaction' and \"session_input\".\"promoted_seq\" is null",
"origin": "manual",
"name": "session_input_session_pending_compaction_idx",
"entityType": "indexes",
"table": "session_input"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "admitted_seq",
"isExpression": false
}
],
"isUnique": true,
"where": null,
"origin": "manual",
"name": "session_input_session_admitted_seq_idx",
"entityType": "indexes",
"table": "session_input"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "promoted_seq",
"isExpression": false
}
],
"isUnique": true,
"where": null,
"origin": "manual",
"name": "session_input_session_promoted_seq_idx",
"entityType": "indexes",
"table": "session_input"
},
{
"columns": [
{
@ -2054,6 +1968,60 @@
"entityType": "indexes",
"table": "session_message"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "delivery",
"isExpression": false
},
{
"value": "admitted_seq",
"isExpression": false
}
],
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_pending_session_delivery_seq_idx",
"entityType": "indexes",
"table": "session_pending"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
}
],
"isUnique": true,
"where": "\"session_pending\".\"type\" = 'compaction'",
"origin": "manual",
"name": "session_pending_session_compaction_idx",
"entityType": "indexes",
"table": "session_pending"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "admitted_seq",
"isExpression": false
}
],
"isUnique": true,
"where": null,
"origin": "manual",
"name": "session_pending_session_admitted_seq_idx",
"entityType": "indexes",
"table": "session_pending"
},
{
"columns": [
{
@ -2111,5 +2079,5 @@
"table": "session"
}
],
"renames": []
"renames": ["session_input->session_pending"]
}

8
packages/core/src/database/database.ts
View file

@ -54,4 +54,10 @@ export function path() {
return join(Global.Path.data, `opencode-${InstallationChannel.replace(/[^a-zA-Z0-9._-]/g, "-")}.db`)
}
export const node = makeGlobalNode({ service: Service, layer: layerFromPath(path()), deps: [] })
// Resolve the database path lazily so tests and embedders that set
// Flag.OPENCODE_DB after module evaluation still control the storage target.
export const node = makeGlobalNode({
service: Service,
layer: Layer.suspend(() => layerFromPath(path())),
deps: [],
})

1
packages/core/src/database/migration.gen.ts generated
View file

@ -52,5 +52,6 @@ export const migrations = (
import("./migration/20260709013000_generic_session_input"),
import("./migration/20260709025533_drop-todo"),
import("./migration/20260709163752_time_suspended"),
import("./migration/20260709190621_session_pending_table"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

35
packages/core/src/database/migration/20260709190621_session_pending_table.ts Normal file
View file

@ -0,0 +1,35 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260709190621_session_pending_table",
up(tx) {
return Effect.gen(function* () {
// Beta reset: session_input becomes the pending-only session_pending
// table. Dropping the old table discards consumed ledger rows and any
// in-flight pending work along with every historical index variant.
yield* tx.run(`DROP TABLE \`session_input\`;`)
yield* tx.run(`
CREATE TABLE \`session_pending\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
\`delivery\` text,
\`admitted_seq\` integer NOT NULL,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_pending_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(
`CREATE INDEX \`session_pending_session_delivery_seq_idx\` ON \`session_pending\` (\`session_id\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_pending_session_compaction_idx\` ON \`session_pending\` (\`session_id\`) WHERE "session_pending"."type" = 'compaction';`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_pending_session_admitted_seq_idx\` ON \`session_pending\` (\`session_id\`,\`admitted_seq\`);`,
)
})
},
} satisfies DatabaseMigration.Migration

46
packages/core/src/database/schema.gen.ts
View file

@ -166,19 +166,6 @@ export default {
CONSTRAINT \`fk_part_message_id_message_id_fk\` FOREIGN KEY (\`message_id\`) REFERENCES \`message\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_input\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
\`delivery\` text,
\`admitted_seq\` integer NOT NULL,
\`promoted_seq\` integer,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_input_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_message\` (
\`id\` text PRIMARY KEY,
@ -191,6 +178,18 @@ export default {
CONSTRAINT \`fk_session_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_pending\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
\`delivery\` text,
\`admitted_seq\` integer NOT NULL,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_pending_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session\` (
\`id\` text PRIMARY KEY,
@ -249,18 +248,6 @@ export default {
)
yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`)
yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`)
yield* tx.run(
`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_pending_compaction_idx\` ON \`session_input\` (\`session_id\`) WHERE "session_input"."type" = 'compaction' and "session_input"."promoted_seq" is null;`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_admitted_seq_idx\` ON \`session_input\` (\`session_id\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_promoted_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`,
)
@ -271,6 +258,15 @@ export default {
`CREATE INDEX \`session_message_session_time_created_id_idx\` ON \`session_message\` (\`session_id\`,\`time_created\`,\`id\`);`,
)
yield* tx.run(`CREATE INDEX \`session_message_time_created_idx\` ON \`session_message\` (\`time_created\`);`)
yield* tx.run(
`CREATE INDEX \`session_pending_session_delivery_seq_idx\` ON \`session_pending\` (\`session_id\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_pending_session_compaction_idx\` ON \`session_pending\` (\`session_id\`) WHERE "session_pending"."type" = 'compaction';`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_pending_session_admitted_seq_idx\` ON \`session_pending\` (\`session_id\`,\`admitted_seq\`);`,
)
yield* tx.run(`CREATE INDEX \`session_project_idx\` ON \`session\` (\`project_id\`);`)
yield* tx.run(`CREATE INDEX \`session_workspace_idx\` ON \`session\` (\`workspace_id\`);`)
yield* tx.run(`CREATE INDEX \`session_parent_idx\` ON \`session\` (\`parent_id\`);`)

46
packages/core/src/session.ts
View file

@ -32,7 +32,7 @@ import { makeGlobalNode } from "./effect/app-node"
import { LocationServiceMap } from "./location-service-map"
import { MessageDecodeError } from "./session/error"
import { SessionEvent } from "./session/event"
import { SessionInput } from "./session/input"
import { SessionPending } from "./session/pending"
import { Snapshot } from "./snapshot"
import { SessionRevert } from "./session/revert"
import { Session } from "@opencode-ai/schema/session"
@ -187,6 +187,12 @@ export interface Interface {
readonly context: (
sessionID: SessionSchema.ID,
) => Effect.Effect<SessionMessage.Info[], NotFoundError | MessageDecodeError>
/**
* Durable admitted session work not yet visible in projected history,
* ordered by admission. Includes unpromoted user and synthetic inputs and
* unhandled compaction barriers.
*/
readonly pending: (sessionID: SessionSchema.ID) => Effect.Effect<SessionPending.Info[], NotFoundError>
/**
* Durable, ordered, gap-free session log read. Replays public durable
* session events after the exclusive `after` cursor, emits a `Synced`
@ -216,9 +222,9 @@ export interface Interface {
files?: PromptInput.Prompt["files"]
agents?: PromptInput.Prompt["agents"]
metadata?: Record<string, unknown>
delivery?: SessionInput.Delivery
delivery?: SessionPending.Delivery
resume?: boolean
}) => Effect.Effect<SessionInput.User, NotFoundError | PromptConflictError | AttachmentError>
}) => Effect.Effect<SessionPending.User, NotFoundError | PromptConflictError | AttachmentError>
readonly command: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
@ -228,10 +234,10 @@ export interface Interface {
model?: ModelV2.Ref
files?: PromptInput.Prompt["files"]
agents?: PromptInput.Prompt["agents"]
delivery?: SessionInput.Delivery
delivery?: SessionPending.Delivery
resume?: boolean
}) => Effect.Effect<
SessionInput.User,
SessionPending.User,
NotFoundError | PromptConflictError | AttachmentError | CommandV2.NotFoundError | CommandV2.EvaluationError
>
readonly shell: (input: {
@ -247,7 +253,7 @@ export interface Interface {
}) => Effect.Effect<void, NotFoundError | SkillNotFoundError>
readonly compact: (
input: CompactInput,
) => Effect.Effect<SessionInput.Compaction, NotFoundError | CompactionConflictError>
) => Effect.Effect<SessionPending.Compaction, NotFoundError | CompactionConflictError>
readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
readonly background: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
@ -259,9 +265,9 @@ export interface Interface {
text: string
description?: string
metadata?: Record<string, unknown>
delivery?: SessionInput.Delivery
delivery?: SessionPending.Delivery
resume?: boolean
}) => Effect.Effect<SessionInput.Synthetic, NotFoundError | SyntheticConflictError>
}) => Effect.Effect<SessionPending.Synthetic, NotFoundError | SyntheticConflictError>
readonly revert: {
readonly stage: (input: {
sessionID: SessionSchema.ID
@ -481,6 +487,10 @@ const layer = Layer.effect(
yield* result.get(sessionID)
return yield* store.context(sessionID)
}),
pending: Effect.fn("V2Session.pending")(function* (sessionID) {
yield* result.get(sessionID)
return yield* SessionPending.list(db, sessionID)
}),
log: (input) =>
Stream.unwrap(
result
@ -504,25 +514,25 @@ const layer = Layer.effect(
Effect.provideService(FSUtil.Service, fs),
)
const messageID = input.id ?? SessionMessage.ID.create()
const admittedInput = SessionInput.Message.make({
const admittedInput = SessionPending.Message.make({
type: "user",
data: { ...prompt, metadata: input.metadata },
delivery: input.delivery ?? "steer",
})
const admitted = yield* SessionInput.admit(db, events, {
const admitted = yield* SessionPending.admit(db, events, {
id: messageID,
sessionID: input.sessionID,
input: admittedInput,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
defect instanceof SessionPending.LifecycleConflict
? new PromptConflictError({ sessionID: input.sessionID, messageID })
: Effect.die(defect),
),
)
if (
admitted.type !== "user" ||
!SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
!SessionPending.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
)
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
if (input.resume !== false) {
@ -664,12 +674,12 @@ const layer = Layer.effect(
compact: Effect.fn("V2Session.compact")(function* (input) {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admitted = yield* SessionInput.admitCompaction(db, events, {
const admitted = yield* SessionPending.admitCompaction(db, events, {
id: inputID,
sessionID: input.sessionID,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
defect instanceof SessionPending.LifecycleConflict
? new CompactionConflictError({ sessionID: input.sessionID, inputID })
: Effect.die(defect),
),
@ -709,7 +719,7 @@ const layer = Layer.effect(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admittedInput = SessionInput.Message.make({
const admittedInput = SessionPending.Message.make({
type: "synthetic",
data: {
text: input.text,
@ -718,20 +728,20 @@ const layer = Layer.effect(
},
delivery: input.delivery ?? "steer",
})
const admitted = yield* SessionInput.admit(db, events, {
const admitted = yield* SessionPending.admit(db, events, {
id: inputID,
sessionID: input.sessionID,
input: admittedInput,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
defect instanceof SessionPending.LifecycleConflict
? new SyntheticConflictError({ sessionID: input.sessionID, inputID })
: Effect.die(defect),
),
)
if (
admitted.type !== "synthetic" ||
!SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
!SessionPending.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
)
return yield* new SyntheticConflictError({ sessionID: input.sessionID, inputID })
if (input.resume !== false && !(yield* result.get(input.sessionID)).revert)

254
packages/core/src/session/input.ts → packages/core/src/session/pending.ts
View file

@ -1,6 +1,6 @@
export * as SessionInput from "./input"
export * as SessionPending from "./pending"
import { and, asc, eq, isNull } from "drizzle-orm"
import { and, asc, eq } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import {
Compaction,
@ -11,14 +11,16 @@ import {
SyntheticData,
User,
UserData,
} from "@opencode-ai/schema/session-input"
} from "@opencode-ai/schema/session-pending"
import { Event } from "@opencode-ai/schema/event"
import type { Database } from "../database/database"
import type { EventV2 } from "../event"
import { EventTable } from "../event/sql"
import { KeyedMutex } from "../effect/keyed-mutex"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { SessionInputTable, SessionMessageTable } from "./sql"
import { SessionMessageTable, SessionPendingTable } from "./sql"
type DatabaseService = Database.Interface["db"]
@ -28,25 +30,28 @@ const decodeUser = Schema.decodeUnknownSync(UserData)
const encodeUser = Schema.encodeSync(UserData)
const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
const encodeSynthetic = Schema.encodeSync(SyntheticData)
const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data)
const admittedEventType = Event.versionedType(
SessionEvent.InputAdmitted.type,
SessionEvent.InputAdmitted.durable.version,
)
const inboxLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()("SessionInput.LifecycleConflict", {
id: SessionMessage.ID,
}) {}
export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()(
"SessionPending.LifecycleConflict",
{
id: SessionMessage.ID,
},
) {}
const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => {
const fromRow = (row: typeof SessionPendingTable.$inferSelect): Info => {
const base = {
admittedSeq: row.admitted_seq,
id: SessionMessage.ID.make(row.id),
sessionID: SessionSchema.ID.make(row.session_id),
timeCreated: DateTime.makeUnsafe(row.time_created),
}
if (row.type === "compaction")
return Compaction.make({
...base,
type: "compaction",
...(row.promoted_seq === null ? {} : { handledSeq: row.promoted_seq }),
})
if (row.type === "compaction") return Compaction.make({ ...base, type: "compaction" })
if (!row.delivery) throw new LifecycleConflict({ id: base.id })
if (row.type === "user")
return User.make({
@ -54,7 +59,6 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => {
type: "user",
data: decodeUser(row.data),
delivery: row.delivery,
...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }),
})
if (row.type === "synthetic")
return Synthetic.make({
@ -62,31 +66,29 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => {
type: "synthetic",
data: decodeSynthetic(row.data),
delivery: row.delivery,
...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }),
})
throw new LifecycleConflict({ id: base.id })
}
export const find = Effect.fn("SessionInput.find")(function* (db: DatabaseService, id: SessionMessage.ID) {
const row = yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie)
export const find = Effect.fn("SessionPending.find")(function* (db: DatabaseService, id: SessionMessage.ID) {
const row = yield* db
.select()
.from(SessionPendingTable)
.where(eq(SessionPendingTable.id, id))
.get()
.pipe(Effect.orDie)
return row === undefined ? undefined : fromRow(row)
})
export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(function* (
export const compaction = Effect.fn("SessionPending.compaction")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const row = yield* db
.select()
.from(SessionInputTable)
.where(
and(
eq(SessionInputTable.session_id, sessionID),
eq(SessionInputTable.type, "compaction"),
isNull(SessionInputTable.promoted_seq),
),
)
.orderBy(asc(SessionInputTable.admitted_seq))
.from(SessionPendingTable)
.where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.type, "compaction")))
.orderBy(asc(SessionPendingTable.admitted_seq))
.limit(1)
.get()
.pipe(Effect.orDie)
@ -95,7 +97,51 @@ export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(fun
return entry.type === "compaction" ? entry : undefined
})
export const admit = Effect.fn("SessionInput.admit")(function* (
/**
* Reconstruct the admitted record for a pending row that was already consumed
* by promotion. The projected `session_message` row proves promotion happened;
* the durable `session.input.admitted` event retains the exact admitted
* message, including delivery.
*/
const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
id: SessionMessage.ID,
) {
const message = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, id))
.get()
.pipe(Effect.orDie)
if (message === undefined) return undefined
if (message.session_id !== sessionID || (message.type !== "user" && message.type !== "synthetic"))
return yield* Effect.die(new LifecycleConflict({ id }))
const rows = yield* db
.select()
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, admittedEventType)))
.all()
.pipe(Effect.orDie)
for (const row of rows) {
const decoded = decodeAdmittedEvent(row.data)
if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue
const base = {
admittedSeq: row.seq,
id,
sessionID,
timeCreated: DateTime.makeUnsafe(row.created),
}
return decoded.value.input.type === "user"
? User.make({ ...base, ...decoded.value.input })
: Synthetic.make({ ...base, ...decoded.value.input })
}
// A projected message without an admitted event in this aggregate (for
// example fork-copied history) is not a retryable admission.
return yield* Effect.die(new LifecycleConflict({ id }))
})
export const admit = Effect.fn("SessionPending.admit")(function* (
db: DatabaseService,
events: EventV2.Interface,
request: {
@ -109,6 +155,8 @@ export const admit = Effect.fn("SessionInput.admit")(function* (
if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
return existing
}
const promoted = yield* promotedFromHistory(db, request.sessionID, request.id)
if (promoted !== undefined) return promoted
return yield* events
.publish(SessionEvent.InputAdmitted, {
inputID: request.id,
@ -141,7 +189,7 @@ export const admit = Effect.fn("SessionInput.admit")(function* (
)
})
export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(function* (
export const admitCompaction = Effect.fn("SessionPending.admitCompaction")(function* (
db: DatabaseService,
events: EventV2.Interface,
input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID },
@ -153,7 +201,7 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}
const pending = yield* pendingCompaction(db, input.sessionID)
const pending = yield* compaction(db, input.sessionID)
if (pending) return pending
return yield* events
.publish(SessionEvent.Compaction.Admitted, {
@ -164,14 +212,14 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
Effect.flatMap((event) => {
if (event.durable === undefined)
return Effect.die(new Error("Compaction admission event is missing aggregate sequence"))
return pendingCompaction(db, input.sessionID).pipe(
return compaction(db, input.sessionID).pipe(
Effect.flatMap((stored) =>
stored ? Effect.succeed(stored) : Effect.die(new LifecycleConflict({ id: input.id })),
),
)
}),
Effect.catchDefect((defect) =>
pendingCompaction(db, input.sessionID).pipe(
compaction(db, input.sessionID).pipe(
Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))),
),
),
@ -180,7 +228,7 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
)
})
export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(function* (
export const projectAdmitted = Effect.fn("SessionPending.projectAdmitted")(function* (
db: DatabaseService,
request: {
readonly admittedSeq: number
@ -198,7 +246,7 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio
.pipe(Effect.orDie)
if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: request.id }))
const stored = yield* db
.insert(SessionInputTable)
.insert(SessionPendingTable)
.values({
id: request.id,
session_id: request.sessionID,
@ -209,13 +257,13 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio
time_created: DateTime.toEpochMillis(request.timeCreated),
})
.onConflictDoNothing()
.returning({ id: SessionInputTable.id })
.returning({ id: SessionPendingTable.id })
.get()
.pipe(Effect.orDie)
if (!stored) return yield* Effect.die(new LifecycleConflict({ id: request.id }))
})
export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompactionAdmitted")(function* (
export const projectCompactionAdmitted = Effect.fn("SessionPending.projectCompactionAdmitted")(function* (
db: DatabaseService,
input: {
readonly admittedSeq: number
@ -232,7 +280,7 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti
.pipe(Effect.orDie)
if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const stored = yield* db
.insert(SessionInputTable)
.insert(SessionPendingTable)
.values({
id: input.id,
session_id: input.sessionID,
@ -249,84 +297,74 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti
const entry = fromRow(stored)
return entry.type === "compaction" ? entry : yield* Effect.die(new LifecycleConflict({ id: entry.id }))
}
const pending = yield* pendingCompaction(db, input.sessionID)
const pending = yield* compaction(db, input.sessionID)
if (pending) return pending
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
})
export const projectPromoted = Effect.fn("SessionInput.projectPromoted")(function* (
/**
* Consume one pending row at promotion. The row's content feeds the projected
* message insert inside the same event transaction; the deleted row is what
* makes the table pending-only.
*/
export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(function* (
db: DatabaseService,
input: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly promotedSeq: number
},
) {
if (yield* pendingCompaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const updated = yield* db
.update(SessionInputTable)
.set({ promoted_seq: input.promotedSeq })
.where(
and(
eq(SessionInputTable.id, input.id),
eq(SessionInputTable.session_id, input.sessionID),
isNull(SessionInputTable.promoted_seq),
),
)
if (yield* compaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const deleted = yield* db
.delete(SessionPendingTable)
.where(and(eq(SessionPendingTable.id, input.id), eq(SessionPendingTable.session_id, input.sessionID)))
.returning()
.get()
.pipe(Effect.orDie)
const stored = updated ? fromRow(updated) : yield* find(db, input.id)
if (
!stored ||
stored.type === "compaction" ||
stored.sessionID !== input.sessionID ||
stored.promotedSeq !== input.promotedSeq
)
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
if (!deleted) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const stored = fromRow(deleted)
if (stored.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: input.id }))
return stored
})
export const settleCompaction = Effect.fn("SessionInput.settleCompaction")(function* (
export const settleCompaction = Effect.fn("SessionPending.settleCompaction")(function* (
db: DatabaseService,
input: { readonly sessionID: SessionSchema.ID; readonly handledSeq: number },
input: { readonly sessionID: SessionSchema.ID },
) {
const updated = yield* db
.update(SessionInputTable)
.set({ promoted_seq: input.handledSeq })
.where(
and(
eq(SessionInputTable.session_id, input.sessionID),
eq(SessionInputTable.type, "compaction"),
isNull(SessionInputTable.promoted_seq),
),
)
const deleted = yield* db
.delete(SessionPendingTable)
.where(and(eq(SessionPendingTable.session_id, input.sessionID), eq(SessionPendingTable.type, "compaction")))
.returning()
.get()
.pipe(Effect.orDie)
if (updated) {
const stored = fromRow(updated)
if (deleted) {
const stored = fromRow(deleted)
return stored.type === "compaction" ? stored : yield* Effect.die(new LifecycleConflict({ id: stored.id }))
}
return undefined
})
export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
export const list = Effect.fn("SessionPending.list")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const rows = yield* db
.select()
.from(SessionPendingTable)
.where(eq(SessionPendingTable.session_id, sessionID))
.orderBy(asc(SessionPendingTable.admitted_seq))
.all()
.pipe(Effect.orDie)
return rows.map(fromRow)
})
export const has = Effect.fn("SessionPending.has")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
delivery: Delivery,
) {
if (yield* pendingCompaction(db, sessionID)) return false
if (yield* compaction(db, sessionID)) return false
const row = yield* db
.select({ id: SessionInputTable.id })
.from(SessionInputTable)
.where(
and(
eq(SessionInputTable.session_id, sessionID),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, delivery),
),
)
.select({ id: SessionPendingTable.id })
.from(SessionPendingTable)
.where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, delivery)))
.limit(1)
.get()
.pipe(Effect.orDie)
@ -350,15 +388,15 @@ export const equivalent = (
return false
}
const publish = Effect.fn("SessionInput.publish")(function* (
const publish = Effect.fn("SessionPending.publish")(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
rows: ReadonlyArray<typeof SessionInputTable.$inferSelect>,
rows: ReadonlyArray<typeof SessionPendingTable.$inferSelect>,
) {
return yield* inboxLocks.withLock(sessionID)(
Effect.gen(function* () {
if (yield* pendingCompaction(db, sessionID)) return 0
if (yield* compaction(db, sessionID)) return 0
yield* Effect.forEach(
rows,
(row) => {
@ -372,12 +410,8 @@ const publish = Effect.fn("SessionInput.publish")(function* (
.pipe(
Effect.catchDefect((defect) =>
defect instanceof LifecycleConflict
? find(db, entry.id).pipe(
Effect.flatMap((stored) =>
stored?.type !== "compaction" && stored?.promotedSeq !== undefined
? Effect.void
: Effect.die(defect),
),
? promotedFromHistory(db, sessionID, entry.id).pipe(
Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
)
: Effect.die(defect),
),
@ -390,45 +424,33 @@ const publish = Effect.fn("SessionInput.publish")(function* (
)
})
export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
export const promoteSteers = Effect.fn("SessionPending.promoteSteers")(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
) {
if (yield* pendingCompaction(db, sessionID)) return 0
if (yield* compaction(db, sessionID)) return 0
const rows = yield* db
.select()
.from(SessionInputTable)
.where(
and(
eq(SessionInputTable.session_id, sessionID),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, "steer"),
),
)
.orderBy(asc(SessionInputTable.admitted_seq))
.from(SessionPendingTable)
.where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "steer")))
.orderBy(asc(SessionPendingTable.admitted_seq))
.all()
.pipe(Effect.orDie)
return yield* publish(db, events, sessionID, rows)
})
export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(function* (
export const promoteNextQueued = Effect.fn("SessionPending.promoteNextQueued")(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
) {
if (yield* pendingCompaction(db, sessionID)) return false
if (yield* compaction(db, sessionID)) return false
const row = yield* db
.select()
.from(SessionInputTable)
.where(
and(
eq(SessionInputTable.session_id, sessionID),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, "queue"),
),
)
.orderBy(asc(SessionInputTable.admitted_seq))
.from(SessionPendingTable)
.where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "queue")))
.orderBy(asc(SessionPendingTable.admitted_seq))
.limit(1)
.get()
.pipe(Effect.orDie)

44
packages/core/src/session/projector.ts
View file

@ -1,6 +1,6 @@
export * as SessionProjector from "./projector"
import { and, asc, desc, eq, gt, gte, inArray, lt, or, sql } from "drizzle-orm"
import { and, asc, desc, eq, gt, gte, inArray, lt, sql } from "drizzle-orm"
import { DateTime, Effect, Layer, Schema, Stream } from "effect"
import { Database } from "../database/database"
import { EventV2 } from "../event"
@ -11,14 +11,14 @@ import { SessionV1 } from "../v1/session"
import { WorkspaceTable } from "../control-plane/workspace.sql"
import { SessionMessage } from "./message"
import { SessionMessageUpdater } from "./message-updater"
import { SessionInput } from "./input"
import { SessionPending } from "./pending"
import { WorkspaceV2 } from "../workspace"
import { InstructionCheckpoint } from "./instruction-checkpoint"
import {
MessageTable,
PartTable,
InstructionCheckpointTable,
SessionInputTable,
SessionPendingTable,
SessionMessageTable,
SessionTable,
} from "./sql"
@ -293,25 +293,25 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.run()
.pipe(Effect.orDie)
const inputRows = yield* db
const pendingRows = yield* db
.select()
.from(SessionInputTable)
.from(SessionPendingTable)
.where(
and(
eq(SessionInputTable.session_id, event.data.parentID),
eq(SessionPendingTable.session_id, event.data.parentID),
inArray(
SessionInputTable.id,
SessionPendingTable.id,
rows.map((row) => row.id),
),
),
)
.all()
.pipe(Effect.orDie)
if (inputRows.length > 0) {
if (pendingRows.length > 0) {
yield* db
.insert(SessionInputTable)
.insert(SessionPendingTable)
.values(
inputRows.flatMap((row) => {
pendingRows.flatMap((row) => {
const id = idMap.get(row.id)
return id && row.type !== "compaction"
? [
@ -322,7 +322,6 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
data: row.data,
delivery: row.delivery,
admitted_seq: row.admitted_seq,
promoted_seq: row.promoted_seq,
time_created: row.time_created,
},
]
@ -633,10 +632,9 @@ const layer = Layer.effectDiscard(
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
const input = yield* SessionInput.projectPromoted(db, {
const input = yield* SessionPending.projectPromoted(db, {
id: event.data.inputID,
sessionID: event.data.sessionID,
promotedSeq: event.durable.seq,
})
yield* insertMessage(
db,
@ -666,7 +664,7 @@ const layer = Layer.effectDiscard(
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
yield* SessionInput.projectAdmitted(db, {
yield* SessionPending.projectAdmitted(db, {
admittedSeq: event.durable.seq,
id: event.data.inputID,
sessionID: event.data.sessionID,
@ -679,7 +677,7 @@ const layer = Layer.effectDiscard(
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
yield* SessionInput.projectCompactionAdmitted(db, {
yield* SessionPending.projectCompactionAdmitted(db, {
admittedSeq: event.durable.seq,
id: event.data.inputID,
sessionID: event.data.sessionID,
@ -727,10 +725,7 @@ const layer = Layer.effectDiscard(
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
if (event.data.reason === "manual")
yield* SessionInput.settleCompaction(db, {
sessionID: event.data.sessionID,
handledSeq: event.durable.seq,
})
yield* SessionPending.settleCompaction(db, { sessionID: event.data.sessionID })
}),
)
yield* events.project(SessionEvent.Compaction.Failed, (event) =>
@ -739,10 +734,7 @@ const layer = Layer.effectDiscard(
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
if (event.data.reason === "manual")
yield* SessionInput.settleCompaction(db, {
sessionID: event.data.sessionID,
handledSeq: event.durable.seq,
})
yield* SessionPending.settleCompaction(db, { sessionID: event.data.sessionID })
}),
)
yield* events.project(SessionEvent.RevertEvent.Staged, (event) =>
@ -786,11 +778,11 @@ const layer = Layer.effectDiscard(
.run()
.pipe(Effect.orDie)
yield* db
.delete(SessionInputTable)
.delete(SessionPendingTable)
.where(
and(
eq(SessionInputTable.session_id, event.data.sessionID),
or(gte(SessionInputTable.admitted_seq, boundary.seq), gte(SessionInputTable.promoted_seq, boundary.seq)),
eq(SessionPendingTable.session_id, event.data.sessionID),
gte(SessionPendingTable.admitted_seq, boundary.seq),
),
)
.run()

34
packages/core/src/session/runner/llm.ts
View file

@ -34,7 +34,7 @@ import { InstructionCheckpoint } from "../instruction-checkpoint"
import { SessionCompaction } from "../compaction"
import { SessionEvent } from "../event"
import { SessionHistory } from "../history"
import { SessionInput } from "../input"
import { SessionPending } from "../pending"
import { SessionMessage } from "../message"
import { SessionSchema } from "../schema"
import { SessionStore } from "../store"
@ -203,7 +203,7 @@ const layer = Layer.effect(
const attemptStep = Effect.fn("SessionRunner.attemptStep")(function* (
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
promotion: SessionPending.Delivery | undefined,
step: number,
recoverOverflow?: typeof compaction.compactAfterOverflow,
assistantMessageID?: SessionMessage.ID,
@ -226,10 +226,10 @@ const layer = Layer.effect(
let currentStep = step
if (promotion) {
let promoted = 0
if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id)
if (promotion === "steer") promoted = yield* SessionPending.promoteSteers(db, events, session.id)
if (promotion === "queue") {
promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id))
promoted += yield* SessionInput.promoteSteers(db, events, session.id)
promoted += Number(yield* SessionPending.promoteNextQueued(db, events, session.id))
promoted += yield* SessionPending.promoteSteers(db, events, session.id)
}
if (promoted > 0) currentStep = 1
}
@ -285,7 +285,7 @@ const layer = Layer.effect(
const advertisedTools = new Set(hookedRequest.tools.map((tool) => tool.name))
// Automatic compaction completed; rebuild the request from compacted history.
if (
!(yield* SessionInput.pendingCompaction(db, session.id)) &&
!(yield* SessionPending.compaction(db, session.id)) &&
(yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request: hookedRequest }))
)
return { _tag: "RestartAfterCompaction", step: currentStep } as const
@ -550,7 +550,7 @@ const layer = Layer.effect(
const runStep = Effect.fnUntraced(function* (
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
promotion: SessionPending.Delivery | undefined,
step: number,
) {
// Compaction restarts rebuild the request from compacted history without re-promoting.
@ -595,7 +595,7 @@ const layer = Layer.effect(
const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* (
sessionID: SessionSchema.ID,
) {
const pending = yield* SessionInput.pendingCompaction(db, sessionID)
const pending = yield* SessionPending.compaction(db, sessionID)
if (!pending) return false
const session = yield* getSession(sessionID)
return yield* Effect.uninterruptibleMask((restore) =>
@ -611,7 +611,7 @@ const layer = Layer.effect(
).pipe(Effect.exit)
if (Exit.isSuccess(compacted) && compacted.value) return true
if (Exit.isFailure(compacted)) {
const unsettled = yield* SessionInput.pendingCompaction(db, sessionID)
const unsettled = yield* SessionPending.compaction(db, sessionID)
if (unsettled)
yield* events.publish(SessionEvent.Compaction.Failed, {
sessionID,
@ -621,7 +621,7 @@ const layer = Layer.effect(
})
return yield* Effect.failCause(compacted.cause)
}
const unsettled = yield* SessionInput.pendingCompaction(db, sessionID)
const unsettled = yield* SessionPending.compaction(db, sessionID)
if (unsettled)
yield* events.publish(SessionEvent.Compaction.Failed, {
sessionID,
@ -640,11 +640,11 @@ const layer = Layer.effect(
readonly force: boolean
}) {
yield* runPendingCompaction(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue")
if (!input.force && !hasSteer && !hasQueue) return
yield* failInterruptedTools(input.sessionID)
let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let promotion: SessionPending.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let shouldRun = input.force || hasSteer || hasQueue
while (shouldRun) {
let needsContinuation = true
@ -664,16 +664,16 @@ const layer = Layer.effect(
needsContinuation = result.needsContinuation
step = result.step + 1
if (needsContinuation) {
promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer"
promotion = (yield* SessionPending.compaction(db, input.sessionID)) ? undefined : "steer"
continue
}
yield* runPendingCompaction(input.sessionID)
promotion = "steer"
needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
needsContinuation = yield* SessionPending.has(db, input.sessionID, "steer")
}
yield* runPendingCompaction(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue")
shouldRun = hasSteer || hasQueue
promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined
}

27
packages/core/src/session/sql.ts
View file

@ -3,7 +3,7 @@ import { sql } from "drizzle-orm"
import { directoryColumn, pathColumn } from "../database/path"
import { ProjectTable } from "../project/sql"
import type { SessionMessage } from "./message"
import type { SessionInput } from "./input"
import type { SessionPending } from "./pending"
import type { FileDiff } from "@opencode-ai/schema/file-diff"
import { PermissionV1 } from "../v1/permission"
import { ProjectV2 } from "../project"
@ -13,7 +13,7 @@ import { WorkspaceV2 } from "../workspace"
import { Timestamps } from "../database/schema.sql"
import type { Instructions } from "../instructions/index"
import type { Session } from "@opencode-ai/schema/session"
import type { SyntheticData, UserData } from "@opencode-ai/schema/session-input"
import type { SyntheticData, UserData } from "@opencode-ai/schema/session-pending"
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
import type { Schema } from "effect"
@ -126,35 +126,28 @@ export const SessionMessageTable = sqliteTable(
],
)
export const SessionInputTable = sqliteTable(
"session_input",
export const SessionPendingTable = sqliteTable(
"session_pending",
{
id: text().$type<SessionMessage.ID>().primaryKey(),
session_id: text()
.$type<SessionSchema.ID>()
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
type: text().$type<SessionInput.Info["type"]>().notNull(),
type: text().$type<SessionPending.Info["type"]>().notNull(),
data: text({ mode: "json" }).$type<UserData | SyntheticData | Record<string, never>>().notNull(),
delivery: text().$type<SessionInput.Delivery>(),
delivery: text().$type<SessionPending.Delivery>(),
admitted_seq: integer().notNull(),
promoted_seq: integer(),
time_created: integer()
.notNull()
.$default(() => Date.now()),
},
(table) => [
index("session_input_session_pending_delivery_seq_idx").on(
table.session_id,
table.promoted_seq,
table.delivery,
table.admitted_seq,
),
uniqueIndex("session_input_session_pending_compaction_idx")
index("session_pending_session_delivery_seq_idx").on(table.session_id, table.delivery, table.admitted_seq),
uniqueIndex("session_pending_session_compaction_idx")
.on(table.session_id)
.where(sql`${table.type} = 'compaction' and ${table.promoted_seq} is null`),
uniqueIndex("session_input_session_admitted_seq_idx").on(table.session_id, table.admitted_seq),
uniqueIndex("session_input_session_promoted_seq_idx").on(table.session_id, table.promoted_seq),
.where(sql`${table.type} = 'compaction'`),
uniqueIndex("session_pending_session_admitted_seq_idx").on(table.session_id, table.admitted_seq),
],
)

82
packages/core/test/database-migration.test.ts
View file

@ -11,14 +11,15 @@ import { migrations } from "@opencode-ai/core/database/migration.gen"
import sessionUsageMigration from "@opencode-ai/core/database/migration/20260510033149_session_usage"
import normalizeStoragePathsMigration from "@opencode-ai/core/database/migration/20260601010001_normalize_storage_paths"
import sessionMessageProjectionOrderMigration from "@opencode-ai/core/database/migration/20260603040000_session_message_projection_order"
import eventSourcedSessionInputMigration from "@opencode-ai/core/database/migration/20260604172448_event_sourced_session_input"
import eventSourcedSessionPendingMigration from "@opencode-ai/core/database/migration/20260604172448_event_sourced_session_input"
import contextEpochAgentMigration from "@opencode-ai/core/database/migration/20260605042240_add_context_epoch_agent"
import simplifyIntegrationCredentialsMigration from "@opencode-ai/core/database/migration/20260611192811_lush_chimera"
import simplifySessionInputMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input"
import simplifySessionPendingMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input"
import resetSessionEventsMigration from "@opencode-ai/core/database/migration/20260703200000_reset_v2_session_events"
import durableSessionInboxMigration from "@opencode-ai/core/database/migration/20260707010146_durable_session_inbox"
import migratePrelaunchV2StateMigration from "@opencode-ai/core/database/migration/20260707120000_migrate_prelaunch_v2_state"
import genericSessionInputMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input"
import genericSessionPendingMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input"
import sessionPendingTableMigration from "@opencode-ai/core/database/migration/20260709190621_session_pending_table"
import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions"
import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork"
import timeSuspendedMigration from "@opencode-ai/core/database/migration/20260709163752_time_suspended"
@ -352,7 +353,10 @@ describe("DatabaseMigration", () => {
})
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`),
).toEqual({ name: "session_input" })
).toBeUndefined()
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_pending'`),
).toEqual({ name: "session_pending" })
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'instruction_checkpoint'`),
).toEqual({ name: "instruction_checkpoint" })
@ -364,18 +368,17 @@ describe("DatabaseMigration", () => {
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: migrations.length })
expect(
yield* db.all(
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_pending_session_delivery_seq_idx', 'session_pending_session_compaction_idx', 'session_pending_session_admitted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
),
).toEqual([
{ name: "event_aggregate_seq_idx" },
{ name: "event_aggregate_type_seq_idx" },
{ name: "session_input_session_admitted_seq_idx" },
{ name: "session_input_session_pending_compaction_idx" },
{ name: "session_input_session_pending_delivery_seq_idx" },
{ name: "session_input_session_promoted_seq_idx" },
{ name: "session_message_session_seq_idx" },
{ name: "session_message_session_time_created_id_idx" },
{ name: "session_message_session_type_seq_idx" },
{ name: "session_pending_session_admitted_seq_idx" },
{ name: "session_pending_session_compaction_idx" },
{ name: "session_pending_session_delivery_seq_idx" },
])
}),
)
@ -568,7 +571,7 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO session_input (id, session_id, prompt, delivery, time_created) VALUES ('msg_pending', 'session', '{}', 'steer', 1)`,
)
yield* DatabaseMigration.applyOnly(db, [eventSourcedSessionInputMigration])
yield* DatabaseMigration.applyOnly(db, [eventSourcedSessionPendingMigration])
expect(yield* db.all(sql`SELECT id, workspace_id FROM session`)).toEqual([
{ id: "session", workspace_id: null },
@ -631,7 +634,7 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO event (id, aggregate_id, seq, type, data, created) VALUES ('event', 'session', 9, 'session.updated.1', '{}', 1)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`,
sql`INSERT INTO session_pending (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`,
)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('projected', 'session', 'user', 9, 1, 1, '{}')`,
@ -640,9 +643,17 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO instruction_checkpoint (session_id, baseline, snapshot, baseline_seq) VALUES ('session', 'baseline', '{}', 9)`,
)
yield* db.run(sql`ALTER TABLE instruction_checkpoint RENAME TO session_context_epoch`)
yield* db.run(sql`DELETE FROM migration WHERE id = ${simplifySessionInputMigration.id}`)
yield* DatabaseMigration.applyOnly(db, [simplifySessionInputMigration])
// The partial compaction index embeds the qualified table name, so it
// must drop before the historical rename dance and recreate after.
yield* db.run(sql`DROP INDEX session_pending_session_compaction_idx`)
yield* db.run(sql`ALTER TABLE session_pending RENAME TO session_input`)
yield* db.run(sql`DELETE FROM migration WHERE id = ${simplifySessionPendingMigration.id}`)
yield* DatabaseMigration.applyOnly(db, [simplifySessionPendingMigration])
yield* db.run(sql`ALTER TABLE session_context_epoch RENAME TO instruction_checkpoint`)
yield* db.run(sql`ALTER TABLE session_input RENAME TO session_pending`)
yield* db.run(
sql`CREATE UNIQUE INDEX session_pending_session_compaction_idx ON session_pending (session_id) WHERE "session_pending"."type" = 'compaction'`,
)
const database = Layer.succeed(Database.Service, { db })
yield* EventV2.Service.use((service) =>
@ -672,7 +683,7 @@ describe("DatabaseMigration", () => {
(SELECT COUNT(*) FROM message WHERE id = 'message') AS messages,
(SELECT COUNT(*) FROM part WHERE id = 'part') AS parts,
(SELECT COUNT(*) FROM workspace) AS workspaces,
(SELECT COUNT(*) FROM session_input) AS sessionInputs,
(SELECT COUNT(*) FROM session_pending) AS sessionInputs,
(SELECT COUNT(*) FROM session_message) AS sessionMessages,
(SELECT COUNT(*) FROM instruction_checkpoint) AS instructionCheckpoints,
(SELECT seq FROM event_sequence WHERE aggregate_id = 'session') AS seq,
@ -756,7 +767,7 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-promoted', 'session', 7, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"empty"}')`,
)
yield* DatabaseMigration.applyOnly(db, [genericSessionInputMigration])
yield* DatabaseMigration.applyOnly(db, [genericSessionPendingMigration])
expect(yield* db.all(sql`SELECT id, type, data, delivery FROM session_input ORDER BY admitted_seq`)).toEqual([
{ id: "input", type: "user", data: '{"text":"hello"}', delivery: "queue" },
@ -775,6 +786,47 @@ describe("DatabaseMigration", () => {
)
})
test("replaces the durable inbox with the empty session_pending table", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
yield* db.run(
sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL REFERENCES session(id) ON DELETE CASCADE, type text NOT NULL, data text NOT NULL, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
)
// Interim v2 builds shipped differing index sets on real databases;
// dropping the table removes whatever variant exists.
yield* db.run(
sql`CREATE INDEX session_input_session_pending_type_delivery_seq_idx ON session_input (session_id, promoted_seq, type, delivery, admitted_seq)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, promoted_seq, time_created) VALUES ('pending', 'session', 'user', '{"text":"hello"}', 'steer', 4, NULL, 1)`,
)
yield* DatabaseMigration.applyOnly(db, [sessionPendingTableMigration])
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`),
).toBeUndefined()
expect(yield* db.all(sql`SELECT id FROM session_pending`)).toEqual([])
expect(
(yield* db.all<{ name: string }>(sql`PRAGMA table_info(session_pending)`)).map((column) => column.name),
).toEqual(["id", "session_id", "type", "data", "delivery", "admitted_seq", "time_created"])
expect(
(yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(session_pending)`))
.filter((index) => index.name.startsWith("session_"))
.map((index) => ({ name: index.name, unique: index.unique }))
.sort((a, b) => a.name.localeCompare(b.name)),
).toEqual([
{ name: "session_pending_session_admitted_seq_idx", unique: 1 },
{ name: "session_pending_session_compaction_idx", unique: 1 },
{ name: "session_pending_session_delivery_seq_idx", unique: 0 },
])
}),
)
})
test("resets incompatible projected Session messages before adding sequence order", async () => {
await run(
Effect.gen(function* () {

4
packages/core/test/session-compact.test.ts
View file

@ -15,7 +15,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session"
import { SessionCompaction } from "@opencode-ai/core/session/compaction"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
@ -106,7 +106,7 @@ describe("SessionV2.compact", () => {
expect(second.id).toBe(first.id)
expect(requests).toHaveLength(0)
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, created.id)).toMatchObject({
expect(yield* SessionPending.compaction((yield* Database.Service).db, created.id)).toMatchObject({
id: first.id,
})
expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toBeUndefined()

52
packages/core/test/session-create.test.ts
View file

@ -20,7 +20,7 @@ import { SessionV1 } from "@opencode-ai/core/v1/session"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
@ -195,9 +195,9 @@ describe("SessionV2.create", () => {
text: "First",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* SessionPending.promoteSteers(db, events, parent.id)
yield* session.synthetic({ sessionID: parent.id, text: "parent note", resume: false })
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* SessionPending.promoteSteers(db, events, parent.id)
const forked = yield* session.fork({ sessionID: parent.id })
const parentContext = yield* session.context(parent.id)
@ -217,30 +217,28 @@ describe("SessionV2.create", () => {
durable: { seq: 0 },
data: { sessionID: forked.id, parentID: parent.id },
})
expect(yield* SessionInput.find(db, forkContext[0].id)).toMatchObject({
sessionID: forked.id,
type: "user",
data: { text: "First" },
promotedSeq: 2,
})
expect(yield* SessionInput.find(db, forkContext[1].id)).toMatchObject({
sessionID: forked.id,
type: "synthetic",
data: { text: "parent note" },
})
expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined()
expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined()
// Fork-copied messages have no admitted event in the fork aggregate, so
// reusing their IDs as prompt IDs is conflicting reuse, not a retry.
expect(
yield* session
.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false })
.pipe(Effect.flip),
).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id })
yield* session.prompt({
sessionID: parent.id,
text: "Parent changed",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* SessionPending.promoteSteers(db, events, parent.id)
yield* session.prompt({
sessionID: forked.id,
text: "Child continues",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, forked.id)
yield* SessionPending.promoteSteers(db, events, forked.id)
expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
@ -250,7 +248,7 @@ describe("SessionV2.create", () => {
(event): number | undefined => event.durable?.seq,
),
).toEqual([0, 5, 6])
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ sessionID: parent.id })
expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
}),
)
@ -265,13 +263,13 @@ describe("SessionV2.create", () => {
text: "First",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* SessionPending.promoteSteers(db, events, parent.id)
const second = yield* session.prompt({
sessionID: parent.id,
text: "Second",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* SessionPending.promoteSteers(db, events, parent.id)
const assistantMessageID = SessionMessage.ID.create()
const model = ModelV2.Ref.make({ id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") })
yield* events.publish(SessionEvent.Step.Started, {
@ -416,7 +414,7 @@ describe("SessionV2.create", () => {
text: "Hello",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, created.id)
yield* SessionPending.promoteSteers(db, events, created.id)
expect(
Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)),
@ -442,7 +440,7 @@ describe("SessionV2.create", () => {
text: "Replay lifecycle",
resume: false,
})
yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id)
yield* SessionPending.promoteSteers(sourceDb, sourceEvents, created.id)
const serialized = (yield* sourceDb
.select()
.from(EventTable)
@ -480,7 +478,7 @@ describe("SessionV2.create", () => {
expect(yield* store.get(created.id)).toBeUndefined()
expect(yield* events.replayAll(serialized.slice(0, 2))).toBe(created.id)
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({
expect(yield* SessionPending.find(db, admitted.id)).toMatchObject({
id: admitted.id,
sessionID: created.id,
type: "user",
@ -491,15 +489,7 @@ describe("SessionV2.create", () => {
expect(yield* store.context(created.id)).toEqual([])
expect(yield* events.replayAll(serialized.slice(2))).toBe(created.id)
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({
id: admitted.id,
sessionID: created.id,
type: "user",
data: { text: "Replay lifecycle" },
delivery: "steer",
admittedSeq: 1,
promotedSeq: 2,
})
expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
expect(yield* store.context(created.id)).toMatchObject([
{ id: admitted.id, type: "user", text: "Replay lifecycle" },
])

19
packages/core/test/session-projector.test.ts
View file

@ -20,11 +20,11 @@ import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { fromRow } from "@opencode-ai/core/session/info"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { Shell } from "@opencode-ai/schema/shell"
import {
InstructionCheckpointTable,
SessionInputTable,
SessionPendingTable,
SessionMessageTable,
SessionTable,
} from "@opencode-ai/core/session/sql"
@ -77,7 +77,7 @@ describe("SessionProjector", () => {
.run()
const events = yield* EventV2.Service
const inputID = SessionMessage.ID.make("msg_manual_compaction")
yield* SessionInput.admitCompaction(db, events, { id: inputID, sessionID })
yield* SessionPending.admitCompaction(db, events, { id: inputID, sessionID })
yield* events.publish(SessionEvent.Compaction.Failed, {
sessionID,
@ -85,7 +85,7 @@ describe("SessionProjector", () => {
error: { type: "compaction.failed", message: "Auto compaction failed" },
})
expect(yield* SessionInput.pendingCompaction(db, sessionID)).toMatchObject({ id: inputID })
expect(yield* SessionPending.compaction(db, sessionID)).toMatchObject({ id: inputID })
}),
)
@ -316,7 +316,7 @@ describe("SessionProjector", () => {
}).pipe(Effect.provide(sessionsLayer)),
)
it.effect("marks an inbox row promoted with the PromptPromoted event sequence", () =>
it.effect("consumes the pending row and projects the message at promotion", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
@ -338,7 +338,7 @@ describe("SessionProjector", () => {
.pipe(Effect.orDie)
const events = yield* EventV2.Service
const id = SessionMessage.ID.make("msg_admitted")
const admitted = yield* SessionInput.admit(db, events, {
const admitted = yield* SessionPending.admit(db, events, {
id,
sessionID,
input: { type: "user", data: { text: "promote me" }, delivery: "steer" },
@ -351,8 +351,11 @@ describe("SessionProjector", () => {
})
expect(
yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie),
).toMatchObject({ promoted_seq: event.durable?.seq })
yield* db.select().from(SessionPendingTable).where(eq(SessionPendingTable.id, id)).get().pipe(Effect.orDie),
).toBeUndefined()
expect(
yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.id, id)).get().pipe(Effect.orDie),
).toMatchObject({ session_id: sessionID, type: "user", seq: event.durable?.seq })
}),
)

99
packages/core/test/session-prompt.test.ts
View file

@ -21,8 +21,8 @@ import { SessionV2 } from "@opencode-ai/core/session"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { SessionPendingTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
import { testEffect } from "./lib/effect"
@ -81,11 +81,11 @@ const setup = Effect.gen(function* () {
.pipe(Effect.orDie)
})
const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionInput.find(db, id))
const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionPending.find(db, id))
const admittedCount = Database.Service.use(({ db }) =>
db
.select()
.from(SessionInputTable)
.from(SessionPendingTable)
.all()
.pipe(
Effect.orDie,
@ -201,7 +201,7 @@ describe("SessionV2.prompt", () => {
text: "boundary",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* SessionPending.promoteSteers(db, events, sessionID)
const stale = SessionMessage.ID.make("msg_stale_assistant")
yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie)
yield* events.publish(SessionEvent.RevertEvent.Staged, {
@ -218,7 +218,7 @@ describe("SessionV2.prompt", () => {
(row) => row.id,
),
).not.toContainAnyValues([boundary.id, stale])
expect(yield* SessionInput.find(db, boundary.id)).toBeUndefined()
expect(yield* SessionPending.find(db, boundary.id)).toBeUndefined()
}),
)
@ -233,7 +233,7 @@ describe("SessionV2.prompt", () => {
text: "boundary",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* SessionPending.promoteSteers(db, events, sessionID)
yield* events.publish(SessionEvent.RevertEvent.Staged, {
sessionID,
revert: { messageID: boundary.id, files: [] },
@ -243,11 +243,11 @@ describe("SessionV2.prompt", () => {
const completion = yield* session.synthetic({ sessionID, text: "stale completion" })
expect(wakeCalls).toEqual([])
expect(yield* SessionInput.find(db, completion.id)).toMatchObject({ type: "synthetic" })
expect(yield* SessionPending.find(db, completion.id)).toMatchObject({ type: "synthetic" })
yield* session.revert.commit(sessionID)
expect(yield* SessionInput.find(db, completion.id)).toBeUndefined()
expect(yield* SessionPending.find(db, completion.id)).toBeUndefined()
}),
)
@ -433,7 +433,7 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({ sessionID, text: "First", resume: false })
yield* session.prompt({ sessionID, text: "Second", resume: false })
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* SessionPending.promoteSteers(db, events, sessionID)
const streamed = Array.from(yield* Fiber.join(fiber))
expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([
@ -610,12 +610,12 @@ describe("SessionV2.prompt", () => {
})
yield* Effect.all(
[SessionInput.promoteSteers(db, events, sessionID), SessionInput.promoteSteers(db, events, sessionID)],
[SessionPending.promoteSteers(db, events, sessionID), SessionPending.promoteSteers(db, events, sessionID)],
{ concurrency: "unbounded" },
)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputPromoted.type, 1))).toBe(1)
expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 })
expect(yield* admitted(messageID)).toBeUndefined()
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: messageID, type: "user", text: "Promote once" },
])
@ -645,7 +645,11 @@ describe("SessionV2.prompt", () => {
.pipe(Effect.orDie)
yield* events.remove(sessionID)
yield* db.delete(SessionInputTable).where(eq(SessionInputTable.session_id, sessionID)).run().pipe(Effect.orDie)
yield* db
.delete(SessionPendingTable)
.where(eq(SessionPendingTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
yield* db
.delete(SessionMessageTable)
.where(eq(SessionMessageTable.session_id, sessionID))
@ -836,7 +840,7 @@ describe("SessionV2.prompt", () => {
},
})
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* SessionPending.promoteSteers(db, events, sessionID)
expect(yield* session.messages({ sessionID })).toMatchObject([
{
@ -861,14 +865,14 @@ describe("SessionV2.prompt", () => {
const entries = yield* Effect.all([session.synthetic(input), session.synthetic(input)], {
concurrency: "unbounded",
})
yield* SessionInput.promoteSteers(database.db, events, sessionID)
yield* SessionPending.promoteSteers(database.db, events, sessionID)
const promotedRetry = yield* session.synthetic(input)
const failure = yield* session.synthetic({ ...input, text: "Different completion" }).pipe(Effect.flip)
expect(entries[1]).toEqual(entries[0])
expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", promotedSeq: expect.any(Number) })
expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", data: { text: "Completed" } })
expect(failure).toMatchObject({ _tag: "Session.SyntheticConflictError", sessionID, inputID: messageID })
expect(yield* admittedCount).toBe(1)
expect(yield* admittedCount).toBe(0)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputAdmitted.type, 1))).toBe(1)
}),
)
@ -888,9 +892,9 @@ describe("SessionV2.prompt", () => {
})
expect(input.delivery).toBe("queue")
expect(yield* SessionInput.promoteSteers(db, events, sessionID)).toBe(0)
expect(yield* SessionPending.promoteSteers(db, events, sessionID)).toBe(0)
expect(yield* session.messages({ sessionID })).toEqual([])
expect(yield* SessionInput.promoteNextQueued(db, events, sessionID)).toBe(true)
expect(yield* SessionPending.promoteNextQueued(db, events, sessionID)).toBe(true)
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: input.id, type: "synthetic", text: "Queued completion" },
])
@ -916,7 +920,7 @@ describe("SessionV2.prompt", () => {
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* SessionPending.promoteSteers(db, events, sessionID)
expect(
(yield* session.messages({ sessionID, order: "asc" })).map((message) =>
@ -926,3 +930,58 @@ describe("SessionV2.prompt", () => {
}),
)
})
describe("SessionV2.pending", () => {
it.effect("fails for an unknown session", () =>
Effect.gen(function* () {
const session = yield* SessionV2.Service
expect(yield* session.pending(SessionV2.ID.make("ses_missing")).pipe(Effect.flip)).toMatchObject({
_tag: "Session.NotFoundError",
})
}),
)
it.effect("lists admitted work in admission order until promotion", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const first = yield* session.prompt({ sessionID, text: "First steer", resume: false })
const queued = yield* session.synthetic({
sessionID,
text: "Queued completion",
delivery: "queue",
resume: false,
})
const second = yield* session.prompt({ sessionID, text: "Second steer", resume: false })
expect(yield* session.pending(sessionID)).toMatchObject([
{ id: first.id, type: "user", delivery: "steer" },
{ id: queued.id, type: "synthetic", delivery: "queue" },
{ id: second.id, type: "user", delivery: "steer" },
])
yield* SessionPending.promoteSteers(db, events, sessionID)
expect(yield* session.pending(sessionID)).toMatchObject([{ id: queued.id, type: "synthetic" }])
yield* SessionPending.promoteNextQueued(db, events, sessionID)
expect(yield* session.pending(sessionID)).toEqual([])
}),
)
it.effect("lists an unhandled compaction barrier until it settles", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const { db } = yield* Database.Service
const barrier = yield* session.compact({ sessionID })
expect(yield* session.pending(sessionID)).toMatchObject([{ id: barrier.id, type: "compaction" }])
yield* SessionPending.settleCompaction(db, { sessionID })
expect(yield* session.pending(sessionID)).toEqual([])
}),
)
})

44
packages/core/test/session-runner.test.ts
View file

@ -27,7 +27,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session"
import { Snapshot } from "@opencode-ai/core/snapshot"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Money } from "@opencode-ai/schema/money"
import { SessionProjector } from "@opencode-ai/core/session/projector"
@ -47,7 +47,7 @@ import { ConfigCompaction } from "@opencode-ai/core/config/compaction"
import { Tool } from "@opencode-ai/core/tool/tool"
import {
InstructionCheckpointTable,
SessionInputTable,
SessionPendingTable,
SessionMessageTable,
SessionTable,
} from "@opencode-ai/core/session/sql"
@ -578,7 +578,7 @@ const replaySessionProjection = (id: SessionV2.ID) =>
.pipe(Effect.orDie)
yield* events.remove(id)
yield* db.delete(SessionInputTable).where(eq(SessionInputTable.session_id, id)).run().pipe(Effect.orDie)
yield* db.delete(SessionPendingTable).where(eq(SessionPendingTable.session_id, id)).run().pipe(Effect.orDie)
yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, id)).run().pipe(Effect.orDie)
yield* events.replayAll(
recorded.map((event) => ({
@ -881,7 +881,7 @@ describe("SessionRunnerLLM", () => {
expect(Exit.isFailure(exit)).toBe(true)
if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(Instructions.InitializationBlocked)
expect(requests).toHaveLength(0)
expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true)
expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true)
expect(
yield* db
.select()
@ -924,7 +924,7 @@ describe("SessionRunnerLLM", () => {
expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
expect(requests).toHaveLength(1)
expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true)
expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true)
}),
)
@ -1462,7 +1462,7 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.compact({ sessionID })
const second = yield* session.compact({ sessionID })
expect(second.id).toBe(first.id)
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toMatchObject({
expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toMatchObject({
id: first.id,
})
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toBeUndefined()
@ -1475,7 +1475,7 @@ describe("SessionRunnerLLM", () => {
delivery: "queue",
resume: false,
})
expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false)
expect(yield* SessionPending.has((yield* Database.Service).db, sessionID, "steer")).toBe(false)
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(active)
@ -1485,7 +1485,7 @@ describe("SessionRunnerLLM", () => {
expect(userTexts(requests[2])).toContain("Steer after compaction")
expect(userTexts(requests[2])).toContain("Completion after compaction")
expect(userTexts(requests[3])).toContain("Queue after compaction")
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({
type: "compaction",
status: "completed",
@ -1521,7 +1521,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(3)
expect(userTexts(requests[2])).toContain("Continue after failure")
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
@ -1542,7 +1542,7 @@ describe("SessionRunnerLLM", () => {
yield* session.resume(sessionID)
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
@ -1566,7 +1566,7 @@ describe("SessionRunnerLLM", () => {
expect(yield* Effect.exit(session.resume(sessionID))).toMatchObject({ _tag: "Failure" })
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect(yield* SessionPending.compaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
@ -2317,7 +2317,7 @@ describe("SessionRunnerLLM", () => {
yield* session.interrupt(sessionID)
expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
expect(requests).toHaveLength(1)
expect(yield* SessionInput.hasPending(db, sessionID, "queue")).toBe(true)
expect(yield* SessionPending.has(db, sessionID, "queue")).toBe(true)
const resumed = yield* session.resume(sessionID).pipe(Effect.forkChild)
while (requests.length < 2) yield* Effect.yieldNow
yield* Deferred.succeed(streamGate, undefined)
@ -2350,7 +2350,7 @@ describe("SessionRunnerLLM", () => {
yield* session.interrupt(sessionID)
expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
expect(requests).toHaveLength(1)
expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true)
expect(yield* SessionPending.has(db, sessionID, "steer")).toBe(true)
const resumed = yield* session.resume(sessionID).pipe(Effect.forkChild)
while (requests.length < 2) yield* Effect.yieldNow
@ -2516,7 +2516,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
const events = yield* EventV2.Service
yield* admit(session, "Recover interrupted tool")
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistantMessageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.Step.Started, {
sessionID,
@ -2573,7 +2573,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
const events = yield* EventV2.Service
yield* admit(session, "Recover interrupted hosted tool")
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistantMessageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.Step.Started, {
sessionID,
@ -2624,7 +2624,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
const events = yield* EventV2.Service
yield* admit(session, "Recover interrupted tool input")
yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID)
yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
const assistantMessageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.Step.Started, {
sessionID,
@ -3056,13 +3056,11 @@ describe("SessionRunnerLLM", () => {
const releaseLateEvent = yield* Deferred.make<void>()
yield* registry.register({ permissionfail: permissionFail })
const events = yield* EventV2.Service
const permissionFailed = yield* events
.subscribe(SessionEvent.Tool.Failed)
.pipe(
Stream.filter((event) => event.data.sessionID === sessionID && event.data.callID === "call-permission"),
Stream.runHead,
Effect.forkScoped({ startImmediately: true }),
)
const permissionFailed = yield* events.subscribe(SessionEvent.Tool.Failed).pipe(
Stream.filter((event) => event.data.sessionID === sessionID && event.data.callID === "call-permission"),
Stream.runHead,
Effect.forkScoped({ startImmediately: true }),
)
yield* admit(session, "Reject permission while another tool input streams")
responseStream = Stream.concat(
Stream.fromIterable([

14
packages/core/test/shared-schema.test.ts
View file

@ -12,7 +12,7 @@ import { ProjectDirectories } from "@opencode-ai/schema/project-directories"
import { PermissionV1 } from "@opencode-ai/schema/permission-v1"
import { Prompt } from "@opencode-ai/schema/prompt"
import { Session } from "@opencode-ai/schema/session"
import { SessionInput } from "@opencode-ai/schema/session-input"
import { SessionPending } from "@opencode-ai/schema/session-pending"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { Workspace } from "@opencode-ai/schema/workspace"
import { Command } from "@opencode-ai/schema/command"
@ -43,7 +43,7 @@ test("Core reuses the canonical shared schemas", async () => {
corePty,
coreProject,
coreReference,
coreSessionInput,
coreSessionPending,
coreSessionMessage,
coreSkill,
coreV2Schema,
@ -63,7 +63,7 @@ test("Core reuses the canonical shared schemas", async () => {
import("@opencode-ai/core/pty"),
import("@opencode-ai/core/project/schema"),
import("@opencode-ai/core/reference"),
import("@opencode-ai/core/session/input"),
import("@opencode-ai/core/session/pending"),
import("@opencode-ai/core/session/message"),
import("@opencode-ai/core/skill"),
import("@opencode-ai/core/v2-schema"),
@ -133,10 +133,10 @@ test("Core reuses the canonical shared schemas", async () => {
[SessionV2.ID, Session.ID],
[SessionV2.Info, Session.Info],
[SessionV2.ListAnchor, Session.ListAnchor],
[coreSessionInput.Delivery, SessionInput.Delivery],
[coreSessionInput.Message, SessionInput.Message],
[coreSessionInput.User, SessionInput.User],
[coreSessionInput.Synthetic, SessionInput.Synthetic],
[coreSessionPending.Delivery, SessionPending.Delivery],
[coreSessionPending.Message, SessionPending.Message],
[coreSessionPending.User, SessionPending.User],
[coreSessionPending.Synthetic, SessionPending.Synthetic],
[coreSessionMessage.ID, SessionMessage.ID],
[coreSessionMessage.AssistantRetry, SessionMessage.AssistantRetry],
[coreSessionMessage.AgentSelected, SessionMessage.AgentSelected],

4
packages/core/test/tool-subagent.test.ts
View file

@ -16,7 +16,7 @@ import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import { SessionV2 } from "@opencode-ai/core/session"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionPending } from "@opencode-ai/core/session/pending"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { SessionStore } from "@opencode-ai/core/session/store"
@ -288,7 +288,7 @@ describe("SubagentTool", () => {
const admission = Array.from(yield* Fiber.join(admitted))[0]
expect(admission?.data.input.data.text).toContain(`<subagent id="${childID}" state="completed"`)
const database = yield* Database.Service
yield* SessionInput.promoteSteers(database.db, events, parent.id)
yield* SessionPending.promoteSteers(database.db, events, parent.id)
const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")
expect(synthetic).toHaveLength(1)
expect(synthetic[0]?.text).toContain(`<subagent id="${childID}" state="completed"`)