Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 60 Packages Projects Releases Wiki Activity

refactor(core): simplify session recovery ownership

Browse Source
This commit is contained in:
Kit Langton
2026-07-07 20:53:42 -04:00
parent c05387ec9c
commit 6c16d4da0f
7 changed files with 95 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/schema.json
+51 -201
View File
@@ -1,10 +1,8 @@
{
"version": "7",
"dialect": "sqlite",
"id": "68b6e591-ccd7-415e-aa5f-94e0ec9da5aa",
"prevIds": [
"b0355fd9-bf41-42e3-9dca-76107de27ecd"
],
"id": "b0355fd9-bf41-42e3-9dca-76107de27ecd",
"prevIds": ["95328a41-789d-44de-9643-6ac6ecd6b4ec"],
"ddl": [
{
"name": "workspace",
@@ -66,10 +64,6 @@
"name": "part",
"entityType": "tables"
},
{
"name": "session_execution_recovery",
"entityType": "tables"
},
{
"name": "session_input",
"entityType": "tables"
@@ -990,26 +984,6 @@
"entityType": "columns",
"table": "part"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "session_id",
"entityType": "columns",
"table": "session_execution_recovery"
},
{
"type": "integer",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "interrupted_seq",
"entityType": "columns",
"table": "session_execution_recovery"
},
{
"type": "text",
"notNull": false,
@@ -1601,13 +1575,9 @@
"table": "session_share"
},
{
"columns": [
"project_id"
],
"columns": ["project_id"],
"tableTo": "project",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1616,13 +1586,9 @@
"table": "workspace"
},
{
"columns": [
"active_account_id"
],
"columns": ["active_account_id"],
"tableTo": "account",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "SET NULL",
"nameExplicit": false,
@@ -1631,13 +1597,9 @@
"table": "account_state"
},
{
"columns": [
"aggregate_id"
],
"columns": ["aggregate_id"],
"tableTo": "event_sequence",
"columnsTo": [
"aggregate_id"
],
"columnsTo": ["aggregate_id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1646,13 +1608,9 @@
"table": "event"
},
{
"columns": [
"project_id"
],
"columns": ["project_id"],
"tableTo": "project",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1661,13 +1619,9 @@
"table": "permission"
},
{
"columns": [
"project_id"
],
"columns": ["project_id"],
"tableTo": "project",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1676,13 +1630,9 @@
"table": "project_directory"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1691,13 +1641,9 @@
"table": "instruction_checkpoint"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1706,13 +1652,9 @@
"table": "instruction_entry"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1721,13 +1663,9 @@
"table": "message"
},
{
"columns": [
"message_id"
],
"columns": ["message_id"],
"tableTo": "message",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1736,28 +1674,9 @@
"table": "part"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
"name": "fk_session_execution_recovery_session_id_session_id_fk",
"entityType": "fks",
"table": "session_execution_recovery"
},
{
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1766,13 +1685,9 @@
"table": "session_input"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1781,13 +1696,9 @@
"table": "session_message"
},
{
"columns": [
"project_id"
],
"columns": ["project_id"],
"tableTo": "project",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1796,13 +1707,9 @@
"table": "session"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1811,13 +1718,9 @@
"table": "todo"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"tableTo": "session",
"columnsTo": [
"id"
],
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@@ -1826,193 +1729,140 @@
"table": "session_share"
},
{
"columns": [
"email",
"url"
],
"columns": ["email", "url"],
"nameExplicit": false,
"name": "control_account_pk",
"entityType": "pks",
"table": "control_account"
},
{
"columns": [
"project_id",
"directory"
],
"columns": ["project_id", "directory"],
"nameExplicit": false,
"name": "project_directory_pk",
"entityType": "pks",
"table": "project_directory"
},
{
"columns": [
"session_id",
"key"
],
"columns": ["session_id", "key"],
"nameExplicit": false,
"name": "instruction_entry_pk",
"entityType": "pks",
"table": "instruction_entry"
},
{
"columns": [
"session_id",
"position"
],
"columns": ["session_id", "position"],
"nameExplicit": false,
"name": "todo_pk",
"entityType": "pks",
"table": "todo"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "workspace_pk",
"table": "workspace",
"entityType": "pks"
},
{
"columns": [
"name"
],
"columns": ["name"],
"nameExplicit": false,
"name": "data_migration_pk",
"table": "data_migration",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "account_state_pk",
"table": "account_state",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "account_pk",
"table": "account",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "credential_pk",
"table": "credential",
"entityType": "pks"
},
{
"columns": [
"aggregate_id"
],
"columns": ["aggregate_id"],
"nameExplicit": false,
"name": "event_sequence_pk",
"table": "event_sequence",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "event_pk",
"table": "event",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "permission_pk",
"table": "permission",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "project_pk",
"table": "project",
"entityType": "pks"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"nameExplicit": false,
"name": "instruction_checkpoint_pk",
"table": "instruction_checkpoint",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "message_pk",
"table": "message",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "part_pk",
"table": "part",
"entityType": "pks"
},
{
"columns": [
"session_id"
],
"nameExplicit": false,
"name": "session_execution_recovery_pk",
"table": "session_execution_recovery",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "session_input_pk",
"table": "session_input",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "session_message_pk",
"table": "session_message",
"entityType": "pks"
},
{
"columns": [
"id"
],
"columns": ["id"],
"nameExplicit": false,
"name": "session_pk",
"table": "session",
"entityType": "pks"
},
{
"columns": [
"session_id"
],
"columns": ["session_id"],
"nameExplicit": false,
"name": "session_share_pk",
"table": "session_share",
@@ -2348,4 +2198,4 @@
}
],
"renames": []
}
}
packages/core/src/database/migration.gen.ts
Generated
-1
View File
@@ -48,6 +48,5 @@ export const migrations = (
import("./migration/20260705180000_rename_instructions"),
import("./migration/20260706223930_add-session-fork"),
import("./migration/20260707010146_durable_session_inbox"),
import("./migration/20260708004136_session-execution-recovery"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]
packages/core/src/database/migration/20260708004136_session-execution-recovery.ts
-17
View File
@@ -1,17 +0,0 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260708004136_session-execution-recovery",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`
CREATE TABLE \`session_execution_recovery\` (
\`session_id\` text PRIMARY KEY,
\`interrupted_seq\` integer NOT NULL,
CONSTRAINT \`fk_session_execution_recovery_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
})
},
} satisfies DatabaseMigration.Migration
packages/core/src/database/schema.gen.ts
-7
View File
@@ -166,13 +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_execution_recovery\` (
\`session_id\` text PRIMARY KEY,
\`interrupted_seq\` integer NOT NULL,
CONSTRAINT \`fk_session_execution_recovery_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_input\` (
\`id\` text PRIMARY KEY,
packages/core/src/session/execution/local.ts
+40 -29
View File
@@ -9,11 +9,11 @@ import { SessionEvent } from "../event"
import { SessionRunCoordinator } from "../run-coordinator"
import { SessionRunner } from "../runner"
import { SessionSchema } from "../schema"
import { SessionExecutionRecoveryTable, SessionTable } from "../sql"
import { SessionStore } from "../store"
import { SessionExecution } from "../execution"
import { toSessionError } from "../to-session-error"
import { UserInterruptedError } from "../error"
import { EffectFlock } from "../../util/effect-flock"
export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?: "user" | "shutdown" | "superseded") {
if (Exit.isSuccess(exit)) return { type: "succeeded" as const }
@@ -23,14 +23,12 @@ export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?:
return { type: "failed" as const, error: toSessionError(failure) }
}
export const claimSessionsInterruptedByShutdown = Effect.fn("SessionExecutionLocal.claimSessionsInterruptedByShutdown")(
export const sessionsInterruptedByShutdown = Effect.fn("SessionExecutionLocal.sessionsInterruptedByShutdown")(
function* (db: Database.Interface["db"]) {
const claimed = yield* db.all<{ sessionID: string }>(sql`
INSERT INTO ${SessionExecutionRecoveryTable} (session_id, interrupted_seq)
SELECT lifecycle.aggregate_id, lifecycle.seq
const latest = yield* db.all<{ sessionID: string; type: string; reason: string | null }>(sql`
SELECT aggregate_id AS sessionID, type, json_extract(data, '$.reason') AS reason
FROM (
SELECT aggregate_id, type, data,
seq,
row_number() OVER (PARTITION BY aggregate_id ORDER BY seq DESC) AS rank
FROM ${EventTable}
WHERE type IN (
@@ -39,16 +37,16 @@ export const claimSessionsInterruptedByShutdown = Effect.fn("SessionExecutionLoc
${EventV2.versionedType(SessionEvent.Execution.Failed.type, 1)},
${EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1)}
)
) AS lifecycle
INNER JOIN ${SessionTable} ON ${SessionTable.id} = lifecycle.aggregate_id
WHERE lifecycle.rank = 1
AND lifecycle.type = ${EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1)}
AND json_extract(lifecycle.data, '$.reason') = 'shutdown'
ON CONFLICT (session_id) DO UPDATE SET interrupted_seq = excluded.interrupted_seq
WHERE interrupted_seq < excluded.interrupted_seq
RETURNING session_id AS sessionID
)
WHERE rank = 1
`)
return claimed.map((event) => SessionSchema.ID.make(event.sessionID))
return latest
.filter(
(event) =>
event.type === EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1) &&
event.reason === "shutdown",
)
.map((event) => SessionSchema.ID.make(event.sessionID))
},
)
@@ -59,6 +57,7 @@ const layer = Layer.effect(
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap.Service
const events = yield* EventV2.Service
const flock = yield* EffectFlock.Service
const { db } = yield* Database.Service
const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
effect.pipe(
@@ -111,19 +110,31 @@ const layer = Layer.effect(
),
})
const interrupted = yield* claimSessionsInterruptedByShutdown(db)
yield* Effect.forEach(
interrupted,
(sessionID) =>
coordinator.run(sessionID).pipe(
Effect.tapCause((cause) =>
Effect.logError("Failed to recover Session after shutdown", cause).pipe(Effect.annotateLogs({ sessionID })),
),
Effect.ignore,
Effect.forkScoped,
),
{ discard: true },
)
yield* flock
.withLock(
Effect.gen(function* () {
const interrupted = yield* sessionsInterruptedByShutdown(db)
yield* Effect.forEach(
interrupted,
(sessionID) =>
coordinator.run(sessionID).pipe(
Effect.tapCause((cause) =>
Effect.logError("Failed to recover Session after shutdown", cause).pipe(
Effect.annotateLogs({ sessionID }),
),
),
Effect.ignore,
),
{ concurrency: "unbounded", discard: true },
)
}),
"session-shutdown-recovery",
)
.pipe(
Effect.tapCause((cause) => Effect.logError("Failed to acquire Session recovery ownership", cause)),
Effect.ignore,
Effect.forkScoped,
)
return SessionExecution.Service.of({
active: coordinator.active,
@@ -138,7 +149,7 @@ const layer = Layer.effect(
export const node = makeGlobalNode({
service: SessionExecution.Service,
layer,
deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, Database.node],
deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, Database.node, EffectFlock.node],
})
export * as SessionExecutionLocal from "./local"
packages/core/src/session/sql.ts
-8
View File
@@ -173,14 +173,6 @@ export const SessionInputTable = sqliteTable(
],
)
export const SessionExecutionRecoveryTable = sqliteTable("session_execution_recovery", {
session_id: text()
.$type<SessionSchema.ID>()
.primaryKey()
.references(() => SessionTable.id, { onDelete: "cascade" }),
interrupted_seq: integer().notNull(),
})
export const InstructionEntryTable = sqliteTable(
"instruction_entry",
{
packages/core/test/session-execution-local.test.ts
+4 -60
View File
@@ -5,14 +5,11 @@ import { EventV2 } from "@opencode-ai/core/event"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { claimSessionsInterruptedByShutdown, terminal } from "@opencode-ai/core/session/execution/local"
import { sessionsInterruptedByShutdown, terminal } from "@opencode-ai/core/session/execution/local"
import { SessionV2 } from "@opencode-ai/core/session"
import { UserInterruptedError } from "@opencode-ai/core/session/error"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
import { sql } from "drizzle-orm"
import { Context, Effect, Exit, Layer } from "effect"
import path from "node:path"
import { tmpdir } from "./fixture/tmpdir"
import { Effect, Exit } from "effect"
import { testEffect } from "./lib/effect"
const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node])))
@@ -46,40 +43,7 @@ describe("SessionExecutionLocal lifecycle", () => {
expect(terminal(Exit.fail(new UserInterruptedError()))).toEqual({ type: "interrupted", reason: "user" })
})
test("claims shutdown recovery once across database connections", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "recovery.sqlite")
await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const first = Context.get(yield* Layer.build(Database.layerFromPath(filename)), Database.Service).db
const second = Context.get(yield* Layer.build(Database.layerFromPath(filename)), Database.Service).db
const sessionID = SessionV2.ID.make("ses_recover_concurrent")
yield* first.run(sql`
INSERT INTO project (id, worktree, sandboxes, time_created, time_updated)
VALUES ('prj_recovery_concurrent', '/tmp/recovery', '[]', 0, 0)
`)
yield* first.run(sql`
INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated)
VALUES (${sessionID}, 'prj_recovery_concurrent', ${sessionID}, '/tmp/recovery', ${sessionID}, 'test', 0, 0)
`)
yield* first.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES (${sessionID}, 1)`)
yield* first.run(sql`
INSERT INTO event (id, aggregate_id, seq, created, type, data)
VALUES ('evt_recovery_concurrent', ${sessionID}, 1, 0, 'session.execution.interrupted.1', '{"sessionID":"ses_recover_concurrent","reason":"shutdown"}')
`)
const claims = yield* Effect.all(
[claimSessionsInterruptedByShutdown(first), claimSessionsInterruptedByShutdown(second)],
{ concurrency: "unbounded" },
)
expect(claims.flat()).toEqual([sessionID])
}),
),
)
})
it.effect("claims each latest shutdown interruption once", () =>
it.effect("selects only sessions whose latest execution ended for shutdown", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const { db } = yield* Database.Service
@@ -88,17 +52,6 @@ describe("SessionExecutionLocal lifecycle", () => {
const user = SessionV2.ID.make("ses_user")
const crashed = SessionV2.ID.make("ses_crashed")
yield* db.run(sql`
INSERT INTO project (id, worktree, sandboxes, time_created, time_updated)
VALUES ('prj_recovery', '/tmp/recovery', '[]', 0, 0)
`)
yield* Effect.forEach([recover, completed, user, crashed], (sessionID) =>
db.run(sql`
INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated)
VALUES (${sessionID}, 'prj_recovery', ${sessionID}, '/tmp/recovery', ${sessionID}, 'test', 0, 0)
`),
)
yield* events.publish(SessionEvent.Execution.Started, { sessionID: recover })
yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: recover, reason: "shutdown" })
yield* events.publish(SessionEvent.InstructionsUpdated, { sessionID: recover, text: "later non-lifecycle event" })
@@ -108,16 +61,7 @@ describe("SessionExecutionLocal lifecycle", () => {
yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: user, reason: "user" })
yield* events.publish(SessionEvent.Execution.Started, { sessionID: crashed })
const claims = yield* Effect.all(
[claimSessionsInterruptedByShutdown(db), claimSessionsInterruptedByShutdown(db)],
{ concurrency: "unbounded" },
)
expect(claims.flat()).toEqual([recover])
yield* events.publish(SessionEvent.Execution.Started, { sessionID: recover })
yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: recover, reason: "shutdown" })
expect(yield* claimSessionsInterruptedByShutdown(db)).toEqual([recover])
expect(yield* claimSessionsInterruptedByShutdown(db)).toEqual([])
expect(yield* sessionsInterruptedByShutdown(db)).toEqual([recover])
}),
)
})