diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 460117b99ab2..9439a461c8ff 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -2232,7 +2232,7 @@ export type SessionInfo = { cost: MoneyUSD tokens: TokenUsageInfo outcome?: "succeeded" | "failed" | "interrupted" - time: { created: number; updated: number; idle?: number; viewed?: number; archived?: number } + time: { created: number; updated: number; busy?: number; idle?: number; viewed?: number; archived?: number } title?: string subpath?: RelativePath metadata?: SessionMetadata @@ -3685,6 +3685,7 @@ export type ExperimentalSessionImportInput = { readonly time: { readonly created: number readonly updated: number + readonly busy?: number readonly idle?: number readonly viewed?: number readonly archived?: number diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index 17c8bbb1227e..30aee4392362 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -1108,6 +1108,10 @@ export function createData(config: CreateDataInput) { return case "session.execution.started": setSessionActive(event.data.sessionID, "running") + // Revalidate so `time.busy` follows the projected execution start. + if (!store.session.info[event.data.sessionID] && !sync.has(`session:${event.data.sessionID}`)) return + result.session.invalidate(event.data.sessionID) + refresh(() => result.session.sync(event.data.sessionID)) return case "session.compaction.started": if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) diff --git a/packages/client/test/solid-data.test.ts b/packages/client/test/solid-data.test.ts index 53d536ca746f..69dea99a2455 100644 --- a/packages/client/test/solid-data.test.ts +++ b/packages/client/test/solid-data.test.ts @@ -330,6 +330,50 @@ test("omits inline tool-result file payloads from read and live tool results whe } }) +test("refreshes a loaded session's execution start when it starts running", async () => { + const listeners = new Set[0]>() + const api = OpenCode.make({ + baseUrl: "http://opencode.local", + fetch: async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init) + if (!request.url.endsWith("/api/session/ses_refresh")) throw new Error(`Unexpected request: ${request.url}`) + return Response.json({ data: { ...session(2), time: { ...session(2).time, busy: 3 } } }) + }, + }) + const event: CreateDataInput["event"] = { + on: + ( + _type: Type, + _handler: (event: Extract) => void, + ) => + () => {}, + listen(handler) { + listeners.add(handler) + return () => listeners.delete(handler) + }, + } + const setup = createRoot((dispose) => ({ + data: createData({ api: () => api, directory: "/project", event, connection: { status: () => "connected" } }), + dispose, + })) + + try { + setup.data.session.remember(session(2)) + const started: OpenCodeEvent = { + id: "evt_started", + created: 3, + type: "session.execution.started", + durable: { aggregateID: "ses_refresh", seq: 1, version: 1 }, + data: { sessionID: "ses_refresh" }, + } + listeners.forEach((listener) => listener({ name: started.type, details: started })) + expect(setup.data.session.status("ses_refresh")).toBe("running") + await wait(() => setup.data.session.get("ses_refresh")?.time.busy === 3) + } finally { + setup.dispose() + } +}) + test("revalidates after an event overtakes an active session read", async () => { let release!: () => void const gate = new Promise((resolve) => (release = resolve)) diff --git a/packages/core/schema.json b/packages/core/schema.json index c643c298402b..bb629130d75f 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "dc16dd12-cb76-432f-8108-0c3aadaf54bb", - "prevIds": ["23940db2-8b4e-4a5e-a404-70d785bd788c"], + "id": "e5daa0af-cae7-4057-919a-ebf44dfced0a", + "prevIds": ["dc16dd12-cb76-432f-8108-0c3aadaf54bb"], "ddl": [ { "name": "account_state", @@ -1672,6 +1672,16 @@ "entityType": "columns", "table": "session_v2" }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_busy", + "entityType": "columns", + "table": "session_v2" + }, { "type": "integer", "notNull": false, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 5438f244a777..f5afbd4fb6d5 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -55,6 +55,7 @@ import m52 from "./migration/20261025123711_blob.js" import m53 from "./migration/20261025154125_drop_session_pending.js" import m54 from "./migration/20261026223005_schedule.js" import m55 from "./migration/20261029002625_session_part.js" +import m56 from "./migration/20261029073502_session_time_busy.js" export const migrations = [ m00, @@ -113,4 +114,5 @@ export const migrations = [ m53, m54, m55, + m56, ] satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20261029073502_session_time_busy.ts b/packages/core/src/database/migration/20261029073502_session_time_busy.ts new file mode 100644 index 000000000000..b0bdd0ee1cfe --- /dev/null +++ b/packages/core/src/database/migration/20261029073502_session_time_busy.ts @@ -0,0 +1,13 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration.js" + +const migration: DatabaseMigration.Migration = { + id: "20261029073502_session_time_busy", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`session_v2\` ADD \`time_busy\` integer;`) + }) + }, +} + +export default migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 6a7ef1601ab1..2cb123ff319b 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -258,6 +258,7 @@ const schema: Omit = { \`model\` text, \`time_created\` integer NOT NULL, \`time_updated\` integer NOT NULL, + \`time_busy\` integer, \`time_idle\` integer, \`time_viewed\` integer, \`idle_outcome\` text, diff --git a/packages/core/src/session/info.ts b/packages/core/src/session/info.ts index ab8a78c6c004..bde75487c249 100644 --- a/packages/core/src/session/info.ts +++ b/packages/core/src/session/info.ts @@ -52,6 +52,7 @@ export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.In time: { created: DateTime.makeUnsafe(row.time_created), updated: DateTime.makeUnsafe(row.time_updated), + busy: row.time_busy === null ? undefined : DateTime.makeUnsafe(row.time_busy), idle: row.time_idle === null ? undefined : DateTime.makeUnsafe(row.time_idle), viewed: row.time_viewed === null ? undefined : DateTime.makeUnsafe(row.time_viewed), archived: row.time_archived === null ? undefined : DateTime.makeUnsafe(row.time_archived), diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index f15c07829f6d..0f0136c1768d 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -787,6 +787,14 @@ const layer = Layer.effectDiscard( delivery: event.data.delivery, }), ) + yield* bus.project(SessionEvent.Execution.Started, (event) => + db + .update(SessionTable) + .set({ time_busy: event.created, time_updated: sql`${SessionTable.time_updated}` }) + .where(eq(SessionTable.id, event.data.sessionID)) + .run() + .pipe(Effect.orDie), + ) yield* bus.project(SessionEvent.Execution.Succeeded, (event) => projectIdle(db, event)) yield* bus.project(SessionEvent.Execution.Failed, (event) => projectIdle(db, event)) yield* bus.project(SessionEvent.Execution.Interrupted, (event) => projectIdle(db, event)) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 0f02d94f38e9..8684925220d9 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -63,6 +63,7 @@ export const SessionTable = sqliteTable( variant?: string }>(), ...Timestamps, + time_busy: integer(), time_idle: integer(), time_viewed: integer(), idle_outcome: text().$type>(), diff --git a/packages/core/src/session/transfer.ts b/packages/core/src/session/transfer.ts index 52e22a57402a..f123a20497e4 100644 --- a/packages/core/src/session/transfer.ts +++ b/packages/core/src/session/transfer.ts @@ -154,6 +154,13 @@ const layer = Layer.effect( tokens_cache_write: input.data.info.tokens.cache.write, time_created: DateTime.toEpochMillis(input.data.info.time.created), time_updated: importedAt, + time_busy: + input.data.info.time.idle && input.data.info.time.busy + ? Math.min( + DateTime.toEpochMillis(input.data.info.time.idle), + DateTime.toEpochMillis(input.data.info.time.busy), + ) + : null, time_idle: input.data.info.time.idle ? DateTime.toEpochMillis(input.data.info.time.idle) : null, time_viewed: input.data.info.time.idle && input.data.info.time.viewed diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 41ed037b9b76..0caa7731b1d3 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -1438,6 +1438,7 @@ describe("SessionTransfer", () => { time: { ...template.time, updated: DateTime.makeUnsafe(100), + busy: DateTime.makeUnsafe(120), idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(150), }, @@ -1479,6 +1480,7 @@ describe("SessionTransfer", () => { }) expect(imported.time).toMatchObject({ updated: DateTime.makeUnsafe(1_000), + busy: DateTime.makeUnsafe(120), idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(150), }) @@ -1488,7 +1490,11 @@ describe("SessionTransfer", () => { ]) expect(yield* Bus.latestSequence(db, sessionID)).toBe(2) const exported = yield* transfer.export({ sessionID }) - expect(exported.info.time).toMatchObject({ idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(150) }) + expect(exported.info.time).toMatchObject({ + busy: DateTime.makeUnsafe(120), + idle: DateTime.makeUnsafe(200), + viewed: DateTime.makeUnsafe(150), + }) expect(exported.messages).toEqual(messages) const sanitized = yield* transfer.export({ sessionID, sanitize: true }) expect(sanitized.info.time).toMatchObject({ idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(150) }) @@ -1527,7 +1533,7 @@ describe("SessionTransfer", () => { }), ) - it.effect("clamps an imported viewed watermark to its idle transition", () => + it.effect("clamps imported busy and viewed watermarks to their idle transition", () => Effect.gen(function* () { const session = yield* Session.Service const transfer = yield* SessionTransfer.Service @@ -1540,6 +1546,7 @@ describe("SessionTransfer", () => { outcome: "succeeded", time: { ...template.time, + busy: DateTime.makeUnsafe(300), idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(250), }, @@ -1549,7 +1556,11 @@ describe("SessionTransfer", () => { location, }) - expect(imported.time).toMatchObject({ idle: DateTime.makeUnsafe(200), viewed: DateTime.makeUnsafe(200) }) + expect(imported.time).toMatchObject({ + busy: DateTime.makeUnsafe(200), + idle: DateTime.makeUnsafe(200), + viewed: DateTime.makeUnsafe(200), + }) expect(imported.outcome).toBe("succeeded") }), ) diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index 53458bbb31f5..ab50f233d935 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -600,6 +600,37 @@ describe("SessionProjector", () => { }), ) + it.effect("records each execution's start, including a resume after shutdown", () => + Effect.gen(function* () { + const db = yield* seedSession({ time_created: 0, time_updated: 0 }) + const bus = yield* Bus.Service + const times = () => + db + .select({ busy: SessionTable.time_busy, idle: SessionTable.time_idle, updated: SessionTable.time_updated }) + .from(SessionTable) + .get() + .pipe(Effect.orDie) + + yield* TestClock.setTime(10) + yield* bus.publish(SessionEvent.Execution.Started, { sessionID }) + expect(yield* times()).toEqual({ busy: 10, idle: null, updated: 0 }) + + yield* TestClock.setTime(20) + yield* bus.publish(SessionEvent.Execution.Interrupted, { sessionID, reason: "shutdown" }) + yield* TestClock.setTime(30) + yield* bus.publish(SessionEvent.Execution.Started, { sessionID }) + expect(yield* times()).toEqual({ busy: 30, idle: null, updated: 0 }) + + yield* TestClock.setTime(40) + yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }) + expect(yield* times()).toEqual({ busy: 30, idle: 40, updated: 0 }) + + yield* TestClock.setTime(50) + yield* bus.publish(SessionEvent.Execution.Started, { sessionID }) + expect(yield* times()).toEqual({ busy: 50, idle: 40, updated: 0 }) + }), + ) + it.effect("updates only the newest incomplete assistant projection", () => Effect.gen(function* () { const db = yield* seedSession() diff --git a/packages/core/test/v1-migration.test.ts b/packages/core/test/v1-migration.test.ts index 2bd411e9a439..adf925d076d6 100644 --- a/packages/core/test/v1-migration.test.ts +++ b/packages/core/test/v1-migration.test.ts @@ -66,6 +66,7 @@ const session = ( model: null, time_created: 1, time_updated: 2, + time_busy: null, time_idle: null, time_viewed: null, idle_outcome: null, diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 3d971b8dd7e8..082023339ecb 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -21856,6 +21856,9 @@ "updated": { "type": "number" }, + "busy": { + "type": "number" + }, "idle": { "type": "number" }, diff --git a/packages/schema/src/session.ts b/packages/schema/src/session.ts index 9f6cd0d71b0c..b03d42cf2965 100644 --- a/packages/schema/src/session.ts +++ b/packages/schema/src/session.ts @@ -46,6 +46,9 @@ export const Info = Schema.Struct({ time: Schema.Struct({ created: DateTimeUtcFromMillis, updated: DateTimeUtcFromMillis, + /** Start of the most recent execution. While it runs, `idle` still holds the previous execution's end. Resuming after a restart starts a new execution. */ + busy: DateTimeUtcFromMillis.pipe(optional), + /** End of the most recent completed execution. */ idle: DateTimeUtcFromMillis.pipe(optional), viewed: DateTimeUtcFromMillis.pipe(optional), archived: DateTimeUtcFromMillis.pipe(optional),