diff --git a/.changeset/execution-history-plugin.md b/.changeset/execution-history-plugin.md new file mode 100644 index 0000000000..ac0f0ab2b0 --- /dev/null +++ b/.changeset/execution-history-plugin.md @@ -0,0 +1,6 @@ +--- +"@executor-js/plugin-execution-history": minor +--- + +Add an optional execution-history plugin that persists execution, tool-call, +and interaction lifecycle events through owner-scoped plugin storage. diff --git a/bun.lock b/bun.lock index 4aebc26736..a109924b96 100644 --- a/bun.lock +++ b/bun.lock @@ -911,6 +911,22 @@ "react", ], }, + "packages/plugins/execution-history": { + "name": "@executor-js/plugin-execution-history", + "version": "0.0.0", + "dependencies": { + "@executor-js/sdk": "workspace:*", + "effect": "catalog:", + }, + "devDependencies": { + "@effect/vitest": "catalog:", + "@types/node": "catalog:", + "bun-types": "catalog:", + "tsup": "catalog:", + "typescript": "catalog:", + "vitest": "catalog:", + }, + }, "packages/plugins/file-secrets": { "name": "@executor-js/plugin-file-secrets", "version": "1.6.0", @@ -1810,6 +1826,8 @@ "@executor-js/plugin-example": ["@executor-js/plugin-example@workspace:packages/plugins/example"], + "@executor-js/plugin-execution-history": ["@executor-js/plugin-execution-history@workspace:packages/plugins/execution-history"], + "@executor-js/plugin-file-secrets": ["@executor-js/plugin-file-secrets@workspace:packages/plugins/file-secrets"], "@executor-js/plugin-graphql": ["@executor-js/plugin-graphql@workspace:packages/plugins/graphql"], diff --git a/packages/plugins/execution-history/CHANGELOG.md b/packages/plugins/execution-history/CHANGELOG.md new file mode 100644 index 0000000000..45c8b65d04 --- /dev/null +++ b/packages/plugins/execution-history/CHANGELOG.md @@ -0,0 +1 @@ +# @executor-js/plugin-execution-history diff --git a/packages/plugins/execution-history/package.json b/packages/plugins/execution-history/package.json new file mode 100644 index 0000000000..f5831010f6 --- /dev/null +++ b/packages/plugins/execution-history/package.json @@ -0,0 +1,28 @@ +{ + "name": "@executor-js/plugin-execution-history", + "version": "0.0.0", + "private": true, + "type": "module", + "exports": { + ".": "./src/sdk/index.ts" + }, + "scripts": { + "build": "tsup && (tsc --declaration --emitDeclarationOnly --outDir dist --rootDir src || true)", + "typecheck": "tsgo --noEmit", + "test": "vitest run", + "test:watch": "vitest", + "typecheck:slow": "tsc --noEmit" + }, + "dependencies": { + "@executor-js/sdk": "workspace:*", + "effect": "catalog:" + }, + "devDependencies": { + "@effect/vitest": "catalog:", + "@types/node": "catalog:", + "bun-types": "catalog:", + "tsup": "catalog:", + "typescript": "catalog:", + "vitest": "catalog:" + } +} diff --git a/packages/plugins/execution-history/src/sdk/collections.ts b/packages/plugins/execution-history/src/sdk/collections.ts new file mode 100644 index 0000000000..9d68ebb2ec --- /dev/null +++ b/packages/plugins/execution-history/src/sdk/collections.ts @@ -0,0 +1,109 @@ +import { Schema } from "effect"; + +import { definePluginStorageCollection } from "@executor-js/sdk/core"; + +// --------------------------------------------------------------------------- +// Execution-history storage collections. +// +// Three owner-scoped plugin-storage collections back the run history: one row +// per execution (`runs`), per tool call (`toolCalls`), and per interaction +// (`interactions`). Every payload that the engine hands us as `unknown` (tool +// args/results, interaction payloads/responses, execution results/logs) is +// stored as an already-serialized JSON string in a `*Json` column so the +// indexed columns stay primitive and query-friendly. Indexes are declared so +// the read surface can filter/sort on them (the facade type-enforces that only +// declared fields appear in `where`/`orderBy`). +// --------------------------------------------------------------------------- + +/** Terminal + transient lifecycle state of a single execution. */ +export const RunStatus = Schema.Literals([ + "running", + "waiting_for_interaction", + "completed", + "failed", +]); +export type RunStatus = typeof RunStatus.Type; + +/** Lifecycle state of a single tool call within an execution. */ +export const ToolCallStatus = Schema.Literals(["running", "completed", "failed"]); +export type ToolCallStatus = typeof ToolCallStatus.Type; + +/** Lifecycle state of a single interaction (elicitation) within an execution. */ +export const InteractionStatus = Schema.Literals([ + "pending", + "accepted", + "declined", + "cancelled", + "failed", +]); +export type InteractionStatus = typeof InteractionStatus.Type; + +export const RunRow = Schema.Struct({ + executionId: Schema.String, + status: RunStatus, + code: Schema.String, + resultJson: Schema.NullOr(Schema.String), + errorText: Schema.NullOr(Schema.String), + logsJson: Schema.NullOr(Schema.String), + triggerKind: Schema.NullOr(Schema.String), + triggerMetaJson: Schema.NullOr(Schema.String), + startedAt: Schema.Number, + completedAt: Schema.NullOr(Schema.Number), + durationMs: Schema.NullOr(Schema.Number), + toolCallCount: Schema.Number, + hadInteraction: Schema.Boolean, +}); +export type RunRow = typeof RunRow.Type; + +export const runs = definePluginStorageCollection("runs", RunRow, { + indexes: ["status", "triggerKind", "startedAt", "durationMs", "hadInteraction"], +}); + +export const ToolCallRow = Schema.Struct({ + executionId: Schema.String, + toolCallId: Schema.String, + status: ToolCallStatus, + path: Schema.String, + namespace: Schema.NullOr(Schema.String), + argsJson: Schema.NullOr(Schema.String), + resultJson: Schema.NullOr(Schema.String), + errorText: Schema.NullOr(Schema.String), + startedAt: Schema.Number, + completedAt: Schema.NullOr(Schema.Number), + durationMs: Schema.NullOr(Schema.Number), +}); +export type ToolCallRow = typeof ToolCallRow.Type; + +export const toolCalls = definePluginStorageCollection("toolCalls", ToolCallRow, { + indexes: ["executionId", "startedAt"], +}); + +export const InteractionRow = Schema.Struct({ + executionId: Schema.String, + interactionId: Schema.String, + status: InteractionStatus, + kind: Schema.String, + purpose: Schema.NullOr(Schema.String), + payloadJson: Schema.NullOr(Schema.String), + responseJson: Schema.NullOr(Schema.String), + errorText: Schema.NullOr(Schema.String), + startedAt: Schema.Number, + completedAt: Schema.NullOr(Schema.Number), +}); +export type InteractionRow = typeof InteractionRow.Type; + +export const interactions = definePluginStorageCollection("interactions", InteractionRow, { + indexes: ["executionId", "startedAt"], +}); + +export const TerminalOutboxRow = Schema.Struct({ + executionId: Schema.String, +}); +export type TerminalOutboxRow = typeof TerminalOutboxRow.Type; + +/** Cleanup markers for terminal outbox blobs. The marker is committed in the + * same batch as the terminal run, then removed after its blob is deleted. */ +export const terminalOutboxes = definePluginStorageCollection( + "terminalOutboxes", + TerminalOutboxRow, +); diff --git a/packages/plugins/execution-history/src/sdk/index.ts b/packages/plugins/execution-history/src/sdk/index.ts new file mode 100644 index 0000000000..f562df4b9b --- /dev/null +++ b/packages/plugins/execution-history/src/sdk/index.ts @@ -0,0 +1,22 @@ +export { executionHistoryPlugin } from "./plugin"; + +export { + interactions, + runs, + toolCalls, + InteractionRow, + InteractionStatus, + RunRow, + RunStatus, + ToolCallRow, + ToolCallStatus, +} from "./collections"; + +export { + makeExecutionHistoryObserver, + makeExecutionHistoryStore, + type ExecutionHistoryDetail, + type ExecutionHistoryListOptions, + type ExecutionHistoryListResult, + type ExecutionHistoryStore, +} from "./store"; diff --git a/packages/plugins/execution-history/src/sdk/plugin.ts b/packages/plugins/execution-history/src/sdk/plugin.ts new file mode 100644 index 0000000000..af8b5dc304 --- /dev/null +++ b/packages/plugins/execution-history/src/sdk/plugin.ts @@ -0,0 +1,32 @@ +import { definePlugin } from "@executor-js/sdk/core"; + +import { interactions, runs, terminalOutboxes, toolCalls } from "./collections"; +import { makeExecutionHistoryObserver, makeExecutionHistoryStore } from "./store"; + +// --------------------------------------------------------------------------- +// Execution-history plugin (SDK surface). A pure sink: it contributes no tools +// or integrations, only the three storage collections, a read surface +// (`list`/`get`/`listToolCalls`) on `executor.executionHistory`, and a runtime +// ExecutionObserver that records the engine's event stream. +// +// One store instance is shared: `storage(deps)` builds it (read methods + +// buffered `handleEvent` writer), `extension(ctx)` surfaces the read methods +// AND `handleEvent` off `ctx.storage`, and `runtime.executionObserver(self)` +// (which receives the EXTENSION) wraps `self.handleEvent` into an observer. +// --------------------------------------------------------------------------- + +export const executionHistoryPlugin = definePlugin(() => ({ + id: "executionHistory" as const, + packageName: "@executor-js/plugin-execution-history", + pluginStorage: { runs, toolCalls, interactions, terminalOutboxes }, + storage: (deps) => makeExecutionHistoryStore(deps), + extension: (ctx) => ({ + list: ctx.storage.list, + get: ctx.storage.get, + listToolCalls: ctx.storage.listToolCalls, + handleEvent: ctx.storage.handleEvent, + }), + runtime: { + executionObserver: (self) => makeExecutionHistoryObserver(self), + }, +})); diff --git a/packages/plugins/execution-history/src/sdk/store.test.ts b/packages/plugins/execution-history/src/sdk/store.test.ts new file mode 100644 index 0000000000..9fd3cf89e4 --- /dev/null +++ b/packages/plugins/execution-history/src/sdk/store.test.ts @@ -0,0 +1,518 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Exit } from "effect"; + +import { + ExecutionFinished, + ExecutionId, + ExecutionInteractionId, + ExecutionStarted, + ExecutionToolCallId, + FormElicitation, + InteractionStarted, + Subject, + Tenant, + ToolAddress, + ToolCallFinished, + ToolCallStarted, + createExecutor, + makeInMemoryBlobStore, + StorageError, +} from "@executor-js/sdk"; +import { makeTestConfig, makeTestExecutor } from "@executor-js/sdk/testing"; + +import { executionHistoryPlugin } from "./plugin"; + +const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; + +describe("execution-history store", () => { + it.effect("records a completed run with one tool call from the event stream", () => + Effect.gen(function* () { + const executor = yield* makeTestExecutor({ + backend: "sqlite", + plugins: [executionHistoryPlugin()] as const, + }); + + const executionId = ExecutionId.make("exec_1"); + const toolCallId = ExecutionToolCallId.make("call_1"); + const startedAt = new Date("2026-05-29T10:00:00.000Z"); + const toolFinishedAt = new Date("2026-05-29T10:00:01.000Z"); + const completedAt = new Date("2026-05-29T10:00:02.000Z"); + + yield* executor.executionHistory.handleEvent( + new ExecutionStarted({ + executionId, + owner, + code: "await tools.shell({ command: 'ls' })", + trigger: { kind: "manual" }, + startedAt, + }), + ); + yield* executor.executionHistory.handleEvent( + new ToolCallStarted({ + executionId, + toolCallId, + owner, + path: "tools.shell.org.default.run", + args: { command: "ls" }, + startedAt, + }), + ); + yield* executor.executionHistory.handleEvent( + new ToolCallFinished({ + executionId, + toolCallId, + owner, + path: "tools.shell.org.default.run", + status: "completed", + result: { stdout: "a.txt" }, + completedAt: toolFinishedAt, + }), + ); + yield* executor.executionHistory.handleEvent( + new ExecutionFinished({ + executionId, + owner, + status: "completed", + result: { ok: true }, + logs: ["ran ls"], + completedAt, + }), + ); + + const listed = yield* executor.executionHistory.list(); + expect(listed.total).toBe(1); + const run = listed.runs[0]; + expect(run?.executionId).toBe("exec_1"); + expect(run?.status).toBe("completed"); + expect(run?.toolCallCount).toBe(1); + expect(run?.durationMs).toBe(2000); + expect(run?.hadInteraction).toBe(false); + // code + trigger from ExecutionStarted survive the terminal re-write. + expect(run?.code).toBe("await tools.shell({ command: 'ls' })"); + expect(run?.triggerKind).toBe("manual"); + + const detail = yield* executor.executionHistory.get("exec_1"); + expect(detail?.run.status).toBe("completed"); + expect(detail?.toolCalls).toHaveLength(1); + expect(detail?.toolCalls[0]?.toolCallId).toBe("call_1"); + expect(detail?.toolCalls[0]?.status).toBe("completed"); + expect(detail?.toolCalls[0]?.durationMs).toBe(1000); + expect(detail?.interactions).toHaveLength(0); + + const toolCallRows = yield* executor.executionHistory.listToolCalls("exec_1"); + expect(toolCallRows).toHaveLength(1); + expect(toolCallRows[0]?.path).toBe("tools.shell.org.default.run"); + }), + ); + + it.effect("keeps waiting history and interaction detail readable after restart", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin()] as const, + }); + const first = yield* createExecutor(config); + + const executionId = ExecutionId.make("exec_waiting"); + const interactionId = ExecutionInteractionId.make("interaction_1"); + const startedAt = new Date("2026-05-29T10:00:00.000Z"); + const interactionAt = new Date("2026-05-29T10:00:01.000Z"); + + yield* first.executionHistory.handleEvent( + new ExecutionStarted({ + executionId, + owner, + code: "await tools.deploy()", + trigger: { kind: "manual" }, + startedAt, + }), + ); + yield* first.executionHistory.handleEvent( + new InteractionStarted({ + executionId, + interactionId, + owner, + context: { + address: ToolAddress.make("tools.deploy.org.default.run"), + args: {}, + request: FormElicitation.make({ + message: "Approve deploy?", + requestedSchema: {}, + }), + }, + startedAt: interactionAt, + }), + ); + + const waiting = yield* first.executionHistory.list(); + expect(waiting.runs[0]?.status).toBe("waiting_for_interaction"); + expect(waiting.runs[0]?.hadInteraction).toBe(true); + const waitingDetail = yield* first.executionHistory.get("exec_waiting"); + expect(waitingDetail?.interactions[0]).toMatchObject({ + interactionId: "interaction_1", + status: "pending", + }); + yield* first.close(); + + const restarted = yield* createExecutor(config); + const detail = yield* restarted.executionHistory.get("exec_waiting"); + expect(detail?.run.status).toBe("waiting_for_interaction"); + expect(detail?.interactions).toHaveLength(1); + expect(detail?.interactions[0]).toMatchObject({ + interactionId: "interaction_1", + status: "pending", + kind: "FormElicitation", + }); + yield* restarted.close(); + yield* Effect.promise(() => config.testDb.close()); + }), + ); + + it.effect( + "recovers a terminal publication after retries are exhausted and the store restarts", + () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin()] as const, + }); + const baseBlobs = makeInMemoryBlobStore(); + let failRecoveryCleanup = false; + const blobs = { + ...baseBlobs, + delete: (namespace: string, key: string) => + failRecoveryCleanup && key === "pending-terminal/exec_recover" + ? Effect.fail( + new StorageError({ + message: "injected recovery cleanup failure", + cause: undefined, + }), + ) + : baseBlobs.delete(namespace, key), + }; + let failInitialRunWrite = true; + let failTerminalWrites = false; + let terminalWriteAttempts = 0; + const withTerminalWriteFault = (source: typeof config.db): typeof config.db => + new Proxy(source, { + get(target, property, receiver) { + if (property === "withContext") { + return (context: unknown) => { + const withContext = target.withContext; + return withContext === undefined + ? target + : withTerminalWriteFault(withContext(context)); + }; + } + if (property === "transaction") { + const transaction: typeof target.transaction = (run) => + target.transaction((transactionDb) => run(withTerminalWriteFault(transactionDb))); + return transaction; + } + if (property === "create") { + const create: typeof target.create = (table, input) => { + if (failInitialRunWrite && table === "plugin_storage") { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB Promise adapter must reject so the SDK maps it into StorageFailure + return Promise.reject( + new StorageError({ + message: "injected initial run write failure", + cause: undefined, + }), + ); + } + return target.create(table, input); + }; + return create; + } + if (property !== "upsertMany") return Reflect.get(target, property, receiver); + return ( + table: Parameters[0], + input: Parameters[1], + ) => { + if (failTerminalWrites && table === "plugin_storage") { + terminalWriteAttempts += 1; + return new Promise((_resolve, reject) => + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB Promise adapter must reject so the SDK maps it into StorageFailure + reject( + new StorageError({ + message: "injected terminal publication failure", + cause: undefined, + }), + ), + ); + } + return target.upsertMany(table, input); + }; + }, + }); + const db = withTerminalWriteFault(config.db); + const executorConfig = { ...config, db, blobs }; + const first = yield* createExecutor(executorConfig); + + const executionId = ExecutionId.make("exec_recover"); + const toolCallId = ExecutionToolCallId.make("call_recover"); + const startedAt = new Date("2026-05-29T10:00:00.000Z"); + const initialWrite = yield* Effect.exit( + first.executionHistory.handleEvent( + new ExecutionStarted({ + executionId, + owner, + code: "await tools.status()", + trigger: { kind: "manual" }, + startedAt, + }), + ), + ); + expect(Exit.isFailure(initialWrite)).toBe(true); + failInitialRunWrite = false; + yield* first.executionHistory.handleEvent( + new ToolCallStarted({ + executionId, + toolCallId, + owner, + path: "tools.status.org.default.get", + args: {}, + startedAt, + }), + ); + yield* first.executionHistory.handleEvent( + new ToolCallFinished({ + executionId, + toolCallId, + owner, + path: "tools.status.org.default.get", + status: "completed", + result: { ok: true }, + completedAt: new Date("2026-05-29T10:00:01.000Z"), + }), + ); + + failTerminalWrites = true; + const failed = yield* Effect.exit( + first.executionHistory.handleEvent( + new ExecutionFinished({ + executionId, + owner, + status: "completed", + result: { ok: true }, + completedAt: new Date("2026-05-29T10:00:02.000Z"), + }), + ), + ); + expect(Exit.isFailure(failed)).toBe(true); + expect(terminalWriteAttempts).toBe(3); + yield* first.close(); + + const restarted = yield* createExecutor(executorConfig); + const visibleWhileReplayFails = yield* restarted.executionHistory.list(); + expect(visibleWhileReplayFails.total).toBe(1); + expect(visibleWhileReplayFails.runs[0]?.status).toBe("running"); + const detailWhileReplayFails = yield* restarted.executionHistory.get("exec_recover"); + expect(detailWhileReplayFails?.run.status).toBe("running"); + const toolCallsWhileReplayFails = + yield* restarted.executionHistory.listToolCalls("exec_recover"); + expect(toolCallsWhileReplayFails).toHaveLength(0); + + failTerminalWrites = false; + failRecoveryCleanup = true; + const recoveredWhileCleanupFails = yield* restarted.executionHistory.get("exec_recover"); + expect(recoveredWhileCleanupFails?.run.status).toBe("completed"); + expect(recoveredWhileCleanupFails?.toolCalls).toHaveLength(1); + const retainedPendingBlob = yield* blobs.has( + "u:test-tenant:test-subject/executionHistory", + "pending-terminal/exec_recover", + ); + expect(retainedPendingBlob).toBe(true); + + failRecoveryCleanup = false; + const completedOnly = yield* restarted.executionHistory.list({ + statusFilter: ["completed"], + }); + expect(completedOnly.total).toBe(1); + expect(completedOnly.runs[0]?.executionId).toBe("exec_recover"); + const detail = yield* restarted.executionHistory.get("exec_recover"); + expect(detail?.run.status).toBe("completed"); + expect(detail?.run.toolCallCount).toBe(1); + expect(detail?.toolCalls).toHaveLength(1); + expect(detail?.toolCalls[0]?.toolCallId).toBe("call_recover"); + const hasPendingBlob = yield* blobs.has( + "u:test-tenant:test-subject/executionHistory", + "pending-terminal/exec_recover", + ); + expect(hasPendingBlob).toBe(false); + yield* restarted.close(); + yield* Effect.promise(() => config.testDb.close()); + }), + ); + + it.effect("publishes directly when a synthetic recovery anchor cannot write its outbox", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin()] as const, + }); + const baseBlobs = makeInMemoryBlobStore(); + let outboxWriteAttempts = 0; + const blobs = { + ...baseBlobs, + put: (namespace: string, key: string, value: string) => { + if (key === "pending-terminal/exec_outbox_unavailable") { + outboxWriteAttempts += 1; + return Effect.fail( + new StorageError({ + message: "injected outbox write failure", + cause: undefined, + }), + ); + } + return baseBlobs.put(namespace, key, value); + }, + }; + let failInitialRunWrite = true; + const withInitialWriteFault = (source: typeof config.db): typeof config.db => + new Proxy(source, { + get(target, property, receiver) { + if (property === "withContext") { + return (context: unknown) => { + const withContext = target.withContext; + return withContext === undefined + ? target + : withInitialWriteFault(withContext(context)); + }; + } + if (property === "transaction") { + const transaction: typeof target.transaction = (run) => + target.transaction((transactionDb) => run(withInitialWriteFault(transactionDb))); + return transaction; + } + if (property !== "create") return Reflect.get(target, property, receiver); + const create: typeof target.create = (table, input) => { + if (failInitialRunWrite && table === "plugin_storage") { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB Promise adapter must reject so the SDK maps it into StorageFailure + return Promise.reject( + new StorageError({ + message: "injected initial run write failure", + cause: undefined, + }), + ); + } + return target.create(table, input); + }; + return create; + }, + }); + const executor = yield* createExecutor({ + ...config, + db: withInitialWriteFault(config.db), + blobs, + }); + const executionId = ExecutionId.make("exec_outbox_unavailable"); + const initialWrite = yield* Effect.exit( + executor.executionHistory.handleEvent( + new ExecutionStarted({ + executionId, + owner, + code: "return true", + trigger: { kind: "manual" }, + startedAt: new Date("2026-05-29T10:00:00.000Z"), + }), + ), + ); + expect(Exit.isFailure(initialWrite)).toBe(true); + failInitialRunWrite = false; + + yield* executor.executionHistory.handleEvent( + new ExecutionFinished({ + executionId, + owner, + status: "completed", + result: true, + completedAt: new Date("2026-05-29T10:00:01.000Z"), + }), + ); + expect(outboxWriteAttempts).toBe(1); + const detail = yield* executor.executionHistory.get("exec_outbox_unavailable"); + expect(detail?.run.status).toBe("completed"); + expect(detail?.run.resultJson).toBe("true"); + const hasPendingBlob = yield* baseBlobs.has( + "u:test-tenant:test-subject/executionHistory", + "pending-terminal/exec_outbox_unavailable", + ); + expect(hasPendingBlob).toBe(false); + yield* executor.close(); + yield* Effect.promise(() => config.testDb.close()); + }), + ); + + it.effect("cleans a published outbox after terminal cleanup exhausts its retries", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin()] as const, + }); + const baseBlobs = makeInMemoryBlobStore(); + let failDeletes = true; + let deleteAttempts = 0; + const blobs = { + ...baseBlobs, + delete: (namespace: string, key: string) => { + if (failDeletes && key === "pending-terminal/exec_cleanup") { + deleteAttempts += 1; + return Effect.fail( + new StorageError({ + message: "injected outbox cleanup failure", + cause: undefined, + }), + ); + } + return baseBlobs.delete(namespace, key); + }, + }; + const executor = yield* createExecutor({ ...config, blobs }); + const executionId = ExecutionId.make("exec_cleanup"); + yield* executor.executionHistory.handleEvent( + new ExecutionStarted({ + executionId, + owner, + code: "return true", + trigger: { kind: "manual" }, + startedAt: new Date("2026-05-29T10:00:00.000Z"), + }), + ); + + yield* executor.executionHistory.handleEvent( + new ExecutionFinished({ + executionId, + owner, + status: "completed", + result: true, + completedAt: new Date("2026-05-29T10:00:01.000Z"), + }), + ); + expect(deleteAttempts).toBe(3); + + const visibleWhileCleanupFails = yield* executor.executionHistory.list({ + statusFilter: ["completed"], + }); + expect(visibleWhileCleanupFails.total).toBe(1); + const retainedPendingBlob = yield* baseBlobs.has( + "u:test-tenant:test-subject/executionHistory", + "pending-terminal/exec_cleanup", + ); + expect(retainedPendingBlob).toBe(true); + + failDeletes = false; + const visibleAfterCleanup = yield* executor.executionHistory.list({ + statusFilter: ["completed"], + }); + expect(visibleAfterCleanup.total).toBe(1); + const hasPendingBlob = yield* baseBlobs.has( + "u:test-tenant:test-subject/executionHistory", + "pending-terminal/exec_cleanup", + ); + expect(hasPendingBlob).toBe(false); + yield* executor.close(); + yield* Effect.promise(() => config.testDb.close()); + }), + ); +}); diff --git a/packages/plugins/execution-history/src/sdk/store.ts b/packages/plugins/execution-history/src/sdk/store.ts new file mode 100644 index 0000000000..d6f59835ee --- /dev/null +++ b/packages/plugins/execution-history/src/sdk/store.ts @@ -0,0 +1,747 @@ +import { Effect, Option, Predicate, Schedule, Schema } from "effect"; + +import { + type ExecutionEvent, + ExecutionInteractionId, + type ExecutionObserver, + ExecutionToolCallId, + type Owner, + type OwnerBinding, + type PluginStorageCollectionFacade, + type PluginStorageFacade, + StorageError, + type StorageDeps, + type StorageFailure, +} from "@executor-js/sdk/core"; + +import { + InteractionRow, + type InteractionStatus, + RunRow, + type RunStatus, + ToolCallRow, + type ToolCallStatus, + interactions, + runs, + terminalOutboxes, + toolCalls, +} from "./collections"; + +// --------------------------------------------------------------------------- +// Execution-history store. Translates the engine's ExecutionEvent stream into +// durable run/tool-call/interaction rows and exposes the read surface. +// +// Write model — buffered batch: tool-call and interaction detail is held in an +// in-memory buffer keyed by executionId and only flushed when the execution +// finishes, so a completed run lands as one batch of writes rather than a +// write-per-event. Before that terminal batch, its complete payload is written +// to the blob seam as a durable outbox; reads replay an unfinished publication +// after a restart. Two points are written eagerly even before the flush: the +// `runs` row on ExecutionStarted (status "running") and again on +// InteractionStarted (status "waiting_for_interaction"), so its history stays +// inspectable if the observer/store restarts while the engine waits on a user. +// +// Every `unknown` payload (tool args/results, interaction payload/response, +// execution result/logs) is serialized to a JSON string via Effect Schema +// (`Schema.UnknownFromJsonString`) — no raw `JSON.stringify` in domain code. +// --------------------------------------------------------------------------- + +/** Serialize an arbitrary value to a JSON string, or null when absent or when + * the value isn't JSON-encodable (encoding never throws). */ +const encodeUnknownJson = Schema.encodeUnknownOption(Schema.UnknownFromJsonString); + +const toJson = (value: unknown): string | null => + value === undefined ? null : Option.getOrNull(encodeUnknownJson(value)); + +const ownerOf = (binding: OwnerBinding): Owner => (binding.subject != null ? "user" : "org"); + +const PendingTerminalPublication = Schema.Struct({ + owner: Schema.Literals(["org", "user"]), + run: RunRow, + toolCalls: Schema.Array(ToolCallRow), + interactions: Schema.Array(InteractionRow), +}); +type PendingTerminalPublication = typeof PendingTerminalPublication.Type; + +const PendingTerminalPublicationFromJsonString = Schema.fromJsonString(PendingTerminalPublication); +const encodePendingTerminalPublication = Schema.encodeUnknownEffect( + PendingTerminalPublicationFromJsonString, +); +const decodePendingTerminalPublication = Schema.decodeUnknownEffect( + PendingTerminalPublicationFromJsonString, +); + +const pendingTerminalBlobKey = (executionId: string): string => `pending-terminal/${executionId}`; + +const isNonterminalRun = (row: RunRow): boolean => + row.status === "running" || row.status === "waiting_for_interaction"; + +/** First dot-delimited segment of a tool path (its namespace), or null. */ +const namespaceOf = (path: string): string | null => { + const index = path.indexOf("."); + return index > 0 ? path.slice(0, index) : null; +}; + +interface BufferedToolCall { + toolCallId: ExecutionToolCallId; + status: ToolCallStatus; + path: string; + namespace: string | null; + argsJson: string | null; + resultJson: string | null; + errorText: string | null; + startedAt: number; + completedAt: number | null; + durationMs: number | null; +} + +interface BufferedInteraction { + interactionId: ExecutionInteractionId; + status: InteractionStatus; + kind: string; + purpose: string | null; + payloadJson: string | null; + responseJson: string | null; + errorText: string | null; + startedAt: number; + completedAt: number | null; +} + +interface RunBuffer { + owner: Owner; + startedAt: number; + // Retained from ExecutionStarted so every re-write of the run row (waiting, + // terminal) keeps the code + trigger — later events don't carry them. + code: string; + triggerKind: string | null; + triggerMetaJson: string | null; + hadInteraction: boolean; + toolCalls: Map; + interactions: Map; +} + +// --------------------------------------------------------------------------- +// Read-surface option/result types. +// --------------------------------------------------------------------------- + +/** Filters and offset pagination for persisted execution summaries. */ +export interface ExecutionHistoryListOptions { + readonly statusFilter?: readonly RunStatus[]; + readonly triggerFilter?: readonly string[]; + readonly timeRange?: { readonly from?: number; readonly to?: number }; + readonly hadInteraction?: boolean; + readonly limit?: number; + readonly offset?: number; + readonly sort?: "asc" | "desc"; +} + +/** A page of execution summaries and the matching row count. */ +export interface ExecutionHistoryListResult { + readonly runs: readonly RunRow[]; + readonly total: number; +} + +/** One execution and its persisted tool-call and interaction records. */ +export interface ExecutionHistoryDetail { + readonly run: RunRow; + readonly toolCalls: readonly ToolCallRow[]; + readonly interactions: readonly InteractionRow[]; +} + +/** Persistence and query capability consumed by the history plugin. */ +export interface ExecutionHistoryStore { + readonly handleEvent: (event: ExecutionEvent) => Effect.Effect; + readonly list: ( + options?: ExecutionHistoryListOptions, + ) => Effect.Effect; + readonly get: ( + executionId: string, + ) => Effect.Effect; + readonly listToolCalls: ( + executionId: string, + ) => Effect.Effect; +} + +/** Create an execution-history store over Executor's owner-scoped plugin storage. */ +export const makeExecutionHistoryStore = (deps: StorageDeps): ExecutionHistoryStore => { + const pluginStorage: PluginStorageFacade = deps.pluginStorage; + const runsC: PluginStorageCollectionFacade = pluginStorage.collection(runs); + const toolCallsC: PluginStorageCollectionFacade = + pluginStorage.collection(toolCalls); + const interactionsC: PluginStorageCollectionFacade = + pluginStorage.collection(interactions); + const terminalOutboxesC: PluginStorageCollectionFacade = + pluginStorage.collection(terminalOutboxes); + const blobs = deps.blobs; + + const buffers = new Map(); + + const putRun = (owner: Owner, row: RunRow): Effect.Effect => + runsC.put({ owner, key: row.executionId, data: row }).pipe(Effect.asVoid); + + const publicationEntries = (publication: PendingTerminalPublication) => [ + ...publication.toolCalls.map((entry) => ({ + collection: toolCalls.name, + key: entry.toolCallId, + data: entry, + })), + ...publication.interactions.map((entry) => ({ + collection: interactions.name, + key: entry.interactionId, + data: entry, + })), + { + collection: runs.name, + key: publication.run.executionId, + data: publication.run, + }, + { + collection: terminalOutboxes.name, + key: publication.run.executionId, + data: { executionId: publication.run.executionId }, + }, + ]; + + const publishTerminal = ( + publication: PendingTerminalPublication, + ): Effect.Effect => + pluginStorage.putMany({ + owner: publication.owner, + entries: publicationEntries(publication), + }); + + const writePendingTerminal = ( + publication: PendingTerminalPublication, + ): Effect.Effect => + encodePendingTerminalPublication(publication).pipe( + Effect.mapError( + (cause) => + new StorageError({ + message: "execution-history: failed to encode pending terminal publication", + cause, + }), + ), + Effect.flatMap((encoded) => + blobs.put(pendingTerminalBlobKey(publication.run.executionId), encoded, { + owner: publication.owner, + }), + ), + ); + + const removePendingTerminal = ( + publication: PendingTerminalPublication, + ): Effect.Effect => + blobs + .delete(pendingTerminalBlobKey(publication.run.executionId), { + owner: publication.owner, + }) + .pipe( + Effect.andThen( + terminalOutboxesC.remove({ + owner: publication.owner, + key: publication.run.executionId, + }), + ), + ); + + const cleanupPublishedTerminal = ( + executionId: string, + owner: Owner, + ): Effect.Effect => + blobs + .delete(pendingTerminalBlobKey(executionId), { owner }) + .pipe(Effect.andThen(terminalOutboxesC.remove({ owner, key: executionId }))); + + const cleanupPublishedTerminals = (): Effect.Effect => + Effect.gen(function* () { + const pendingCleanup = yield* terminalOutboxesC.list(); + yield* Effect.forEach( + pendingCleanup, + (entry) => + cleanupPublishedTerminal(entry.data.executionId, entry.owner).pipe( + Effect.retry(Schedule.recurs(2)), + Effect.ignore, + ), + { concurrency: 4, discard: true }, + ); + }).pipe(Effect.retry(Schedule.recurs(2)), Effect.ignore); + + const cleanupPublishedTerminalIfMarked = (executionId: string): Effect.Effect => + Effect.gen(function* () { + const marker = yield* terminalOutboxesC.get({ key: executionId }); + if (marker !== null) { + yield* cleanupPublishedTerminal(executionId, marker.owner); + } + }).pipe(Effect.retry(Schedule.recurs(2)), Effect.ignore); + + const recoverPendingTerminal = (executionId: string): Effect.Effect => + Effect.gen(function* () { + const encoded = yield* blobs.get(pendingTerminalBlobKey(executionId)); + if (encoded === null) return false; + const publication = yield* decodePendingTerminalPublication(encoded).pipe( + Effect.mapError( + (cause) => + new StorageError({ + message: "execution-history: failed to decode pending terminal publication", + cause, + }), + ), + ); + yield* publishTerminal(publication); + // Publication is the recovery success boundary. Cleanup is idempotent + // and backed by the marker published above, so an unavailable blob store + // must not make a reader return the stale nonterminal row it first read. + yield* removePendingTerminal(publication).pipe( + Effect.retry(Schedule.recurs(2)), + Effect.ignore, + ); + return true; + }).pipe(Effect.retry(Schedule.recurs(2))); + + const recoverOutstandingTerminals = (): Effect.Effect => + Effect.gen(function* () { + const candidates = yield* runsC.query({ + where: { status: { in: ["running", "waiting_for_interaction"] } }, + }); + yield* Effect.forEach( + candidates, + (entry) => recoverPendingTerminal(entry.data.executionId).pipe(Effect.ignore), + { concurrency: 1, discard: true }, + ); + }).pipe(Effect.ignore); + + const toolCallRowsFromBuffer = (executionId: string, buffer: RunBuffer): readonly ToolCallRow[] => + Array.from(buffer.toolCalls.values(), (entry) => ({ + executionId, + toolCallId: entry.toolCallId, + status: entry.status, + path: entry.path, + namespace: entry.namespace, + argsJson: entry.argsJson, + resultJson: entry.resultJson, + errorText: entry.errorText, + startedAt: entry.startedAt, + completedAt: entry.completedAt, + durationMs: entry.durationMs, + })); + + const interactionRowsFromBuffer = ( + executionId: string, + buffer: RunBuffer, + ): readonly InteractionRow[] => + Array.from(buffer.interactions.values(), (entry) => ({ + executionId, + interactionId: entry.interactionId, + status: entry.status, + kind: entry.kind, + purpose: entry.purpose, + payloadJson: entry.payloadJson, + responseJson: entry.responseJson, + errorText: entry.errorText, + startedAt: entry.startedAt, + completedAt: entry.completedAt, + })); + + const getOrLoadBuffer = (executionId: string): Effect.Effect => + Effect.gen(function* () { + const current = buffers.get(executionId); + if (current) return current; + + const run = yield* runsC.get({ key: executionId }); + if (run === null) return null; + const persistedToolCalls = yield* toolCallsC.query({ + where: { executionId }, + orderBy: [{ field: "startedAt" }], + }); + const persistedInteractions = yield* interactionsC.query({ + where: { executionId }, + orderBy: [{ field: "startedAt" }], + }); + const buffer: RunBuffer = { + owner: run.owner, + startedAt: run.data.startedAt, + code: run.data.code, + triggerKind: run.data.triggerKind, + triggerMetaJson: run.data.triggerMetaJson, + hadInteraction: run.data.hadInteraction, + toolCalls: new Map( + persistedToolCalls.map(({ data }) => [ + data.toolCallId, + { + toolCallId: ExecutionToolCallId.make(data.toolCallId), + status: data.status, + path: data.path, + namespace: data.namespace, + argsJson: data.argsJson, + resultJson: data.resultJson, + errorText: data.errorText, + startedAt: data.startedAt, + completedAt: data.completedAt, + durationMs: data.durationMs, + }, + ]), + ), + interactions: new Map( + persistedInteractions.map(({ data }) => [ + data.interactionId, + { + interactionId: ExecutionInteractionId.make(data.interactionId), + status: data.status, + kind: data.kind, + purpose: data.purpose, + payloadJson: data.payloadJson, + responseJson: data.responseJson, + errorText: data.errorText, + startedAt: data.startedAt, + completedAt: data.completedAt, + }, + ]), + ), + }; + buffers.set(executionId, buffer); + return buffer; + }); + + const onExecutionStarted = (event: Extract) => { + const owner = ownerOf(event.owner); + const startedAt = event.startedAt.getTime(); + const triggerKind = event.trigger?.kind ?? null; + const triggerMetaJson = toJson(event.trigger?.metadata); + buffers.set(event.executionId, { + owner, + startedAt, + code: event.code, + triggerKind, + triggerMetaJson, + hadInteraction: false, + toolCalls: new Map(), + interactions: new Map(), + }); + return putRun(owner, { + executionId: event.executionId, + status: "running", + code: event.code, + resultJson: null, + errorText: null, + logsJson: null, + triggerKind, + triggerMetaJson, + startedAt, + completedAt: null, + durationMs: null, + toolCallCount: 0, + hadInteraction: false, + }); + }; + + const onToolCallStarted = (event: Extract) => + Effect.gen(function* () { + const buffer = yield* getOrLoadBuffer(event.executionId); + if (buffer === null) return; + buffer.toolCalls.set(event.toolCallId, { + toolCallId: event.toolCallId, + status: "running", + path: event.path, + namespace: namespaceOf(event.path), + argsJson: toJson(event.args), + resultJson: null, + errorText: null, + startedAt: event.startedAt.getTime(), + completedAt: null, + durationMs: null, + }); + }); + + const onToolCallFinished = (event: Extract) => + Effect.gen(function* () { + const buffer = yield* getOrLoadBuffer(event.executionId); + if (buffer === null) return; + const completedAt = event.completedAt.getTime(); + const existing = buffer.toolCalls.get(event.toolCallId); + const startedAt = existing?.startedAt ?? completedAt; + const row: ToolCallRow = { + executionId: event.executionId, + toolCallId: event.toolCallId, + status: event.status, + path: event.path, + namespace: existing?.namespace ?? namespaceOf(event.path), + argsJson: existing?.argsJson ?? null, + resultJson: toJson(event.result), + errorText: event.error ?? null, + startedAt, + completedAt, + durationMs: completedAt - startedAt, + }; + buffer.toolCalls.set(event.toolCallId, { ...row, toolCallId: event.toolCallId }); + // Once a run has reached a durable waiting point, keep subsequent + // progress restart-safe until the terminal publication lands. + if (buffer.hadInteraction) { + yield* toolCallsC.put({ owner: buffer.owner, key: row.toolCallId, data: row }); + } + }); + + const onInteractionStarted = (event: Extract) => + Effect.gen(function* () { + const buffer = yield* getOrLoadBuffer(event.executionId); + if (buffer === null) return; + const request = event.context.request; + const kind = Predicate.isTagged(request, "UrlElicitation") + ? "UrlElicitation" + : "FormElicitation"; + buffer.interactions.set(event.interactionId, { + interactionId: event.interactionId, + status: "pending", + kind, + purpose: request.message, + payloadJson: toJson(event.context), + responseJson: null, + errorText: null, + startedAt: event.startedAt.getTime(), + completedAt: null, + }); + buffer.hadInteraction = true; + const waitingRun: RunRow = { + executionId: event.executionId, + status: "waiting_for_interaction", + code: buffer.code, + resultJson: null, + errorText: null, + logsJson: null, + triggerKind: buffer.triggerKind, + triggerMetaJson: buffer.triggerMetaJson, + startedAt: buffer.startedAt, + completedAt: null, + durationMs: null, + toolCallCount: buffer.toolCalls.size, + hadInteraction: true, + }; + yield* pluginStorage.putMany({ + owner: buffer.owner, + entries: [ + ...toolCallRowsFromBuffer(event.executionId, buffer).map((row) => ({ + collection: toolCalls.name, + key: row.toolCallId, + data: row, + })), + ...interactionRowsFromBuffer(event.executionId, buffer).map((row) => ({ + collection: interactions.name, + key: row.interactionId, + data: row, + })), + { collection: runs.name, key: event.executionId, data: waitingRun }, + ], + }); + }).pipe(Effect.retry(Schedule.recurs(2))); + + const onInteractionResolved = (event: Extract) => + Effect.gen(function* () { + const buffer = yield* getOrLoadBuffer(event.executionId); + if (buffer === null) return; + const existing = buffer.interactions.get(event.interactionId); + const row: InteractionRow = { + executionId: event.executionId, + interactionId: event.interactionId, + status: event.status, + kind: existing?.kind ?? "unknown", + purpose: existing?.purpose ?? null, + payloadJson: existing?.payloadJson ?? null, + responseJson: toJson(event.response), + errorText: event.error ?? null, + startedAt: existing?.startedAt ?? event.completedAt.getTime(), + completedAt: event.completedAt.getTime(), + }; + buffer.interactions.set(event.interactionId, { + ...row, + interactionId: event.interactionId, + }); + yield* interactionsC.put({ owner: buffer.owner, key: row.interactionId, data: row }); + }); + + const onExecutionFinished = (event: Extract) => + Effect.gen(function* () { + const buffer = yield* getOrLoadBuffer(event.executionId); + const owner = buffer?.owner ?? ownerOf(event.owner); + const completedAt = event.completedAt.getTime(); + const toolCallEntries = buffer ? toolCallRowsFromBuffer(event.executionId, buffer) : []; + const interactionEntries = buffer ? interactionRowsFromBuffer(event.executionId, buffer) : []; + // Preserve code/trigger/startedAt from the buffer, or from the persisted + // "running" row if the buffer was lost (e.g. a restart mid-run). + const existing = yield* runsC.get({ key: event.executionId }); + const code = buffer?.code ?? existing?.data.code ?? ""; + const triggerKind = buffer?.triggerKind ?? existing?.data.triggerKind ?? null; + const triggerMetaJson = buffer?.triggerMetaJson ?? existing?.data.triggerMetaJson ?? null; + const startedAt = buffer?.startedAt ?? existing?.data.startedAt ?? completedAt; + const hadInteraction = + buffer?.hadInteraction ?? (existing?.data.hadInteraction || interactionEntries.length > 0); + + const terminalRun: RunRow = { + executionId: event.executionId, + status: event.status, + code, + resultJson: toJson(event.result), + errorText: event.error ?? null, + logsJson: toJson(event.logs), + triggerKind, + triggerMetaJson, + startedAt, + completedAt, + durationMs: completedAt - startedAt, + toolCallCount: toolCallEntries.length, + hadInteraction, + }; + + const publication: PendingTerminalPublication = { + owner, + run: terminalRun, + toolCalls: toolCallEntries, + interactions: interactionEntries, + }; + + // Observer failures are isolated from the engine. If the eager started + // write failed, establish a nonterminal recovery anchor before creating + // the outbox; reads can then discover and replay that blob after restart. + if (existing === null) { + yield* putRun(owner, { + ...terminalRun, + status: "running", + resultJson: null, + errorText: null, + logsJson: null, + completedAt: null, + durationMs: null, + }); + } + + // The blob is a durable outbox. Once it exists, a later read can replay + // this idempotent publication after a restart or exhausted retry budget. + // If the outbox store is unavailable, continue to the atomic database + // publication directly instead of losing the completed execution. + yield* writePendingTerminal(publication).pipe(Effect.catch(() => Effect.void)); + // One bulk upsert is the terminal commit point: the run and all of its + // normalized detail rows become visible together, so readers can never + // observe a terminal count without the corresponding records. + yield* publishTerminal(publication); + // The terminal batch above is the commit point. Cleanup is retryable via + // its durable marker and must not report a committed history write as a + // failure merely because the blob store is temporarily unavailable. + yield* removePendingTerminal(publication).pipe( + Effect.retry(Schedule.recurs(2)), + Effect.ignore, + ); + }).pipe( + // The observer boundary logs and isolates failures, so retry transient + // storage faults here. Repeated puts are idempotent by collection key. + Effect.retry(Schedule.recurs(2)), + // Keep the live buffer until the durable outbox has been published and + // removed. Once written, that outbox also survives process restarts. + Effect.tap(() => Effect.sync(() => buffers.delete(event.executionId))), + ); + + const handleEvent = (event: ExecutionEvent): Effect.Effect => { + if (Predicate.isTagged(event, "ExecutionStarted")) return onExecutionStarted(event); + if (Predicate.isTagged(event, "ToolCallStarted")) return onToolCallStarted(event); + if (Predicate.isTagged(event, "ToolCallFinished")) return onToolCallFinished(event); + if (Predicate.isTagged(event, "InteractionStarted")) return onInteractionStarted(event); + if (Predicate.isTagged(event, "InteractionResolved")) return onInteractionResolved(event); + // Explicit guard, not a fallthrough: a future ExecutionEvent variant must + // not be silently recorded as a finished run. + if (Predicate.isTagged(event, "ExecutionFinished")) return onExecutionFinished(event); + return Effect.void; + }; + + const list = ( + options?: ExecutionHistoryListOptions, + ): Effect.Effect => { + const where: { + status?: { in: readonly RunStatus[] }; + triggerKind?: { in: readonly string[] }; + startedAt?: { gte?: number; lte?: number }; + hadInteraction?: { eq: boolean }; + } = {}; + if (options?.statusFilter && options.statusFilter.length > 0) { + where.status = { in: options.statusFilter }; + } + if (options?.triggerFilter && options.triggerFilter.length > 0) { + where.triggerKind = { in: options.triggerFilter }; + } + if (options?.timeRange) { + where.startedAt = {}; + if (options.timeRange.from != null) where.startedAt.gte = options.timeRange.from; + if (options.timeRange.to != null) where.startedAt.lte = options.timeRange.to; + } + if (options?.hadInteraction != null) { + where.hadInteraction = { eq: options.hadInteraction }; + } + + return Effect.gen(function* () { + // Drain the durable outbox independently of the caller's filters and + // page so a completed run cannot remain hidden as nonterminal. + yield* cleanupPublishedTerminals(); + yield* recoverOutstandingTerminals(); + const rows = yield* runsC.query({ + where, + orderBy: [{ field: "startedAt", direction: options?.sort ?? "desc" }], + limit: options?.limit, + offset: options?.offset, + }); + const total = yield* runsC.count({ where }); + return { runs: rows.map((entry) => entry.data), total }; + }); + }; + + const get = (executionId: string): Effect.Effect => + Effect.gen(function* () { + let run = yield* runsC.get({ key: executionId }); + if (run === null) return null; + const recovered = isNonterminalRun(run.data) + ? yield* recoverPendingTerminal(executionId).pipe(Effect.orElseSucceed(() => false)) + : false; + if (recovered) { + run = yield* runsC.get({ key: executionId }); + if (run === null) return null; + } + if (!isNonterminalRun(run.data)) { + yield* cleanupPublishedTerminalIfMarked(executionId); + } + const toolCallRows = yield* toolCallsC.query({ + where: { executionId }, + orderBy: [{ field: "startedAt" }], + }); + const interactionRows = yield* interactionsC.query({ + where: { executionId }, + orderBy: [{ field: "startedAt" }], + }); + return { + run: run.data, + toolCalls: toolCallRows.map((entry) => entry.data), + interactions: interactionRows.map((entry) => entry.data), + }; + }); + + const listToolCalls = ( + executionId: string, + ): Effect.Effect => + Effect.gen(function* () { + const run = yield* runsC.get({ key: executionId }); + if (run !== null && isNonterminalRun(run.data)) { + yield* recoverPendingTerminal(executionId).pipe(Effect.ignore); + } else if (run !== null) { + yield* cleanupPublishedTerminalIfMarked(executionId); + } + const rows = yield* toolCallsC.query({ + where: { executionId }, + orderBy: [{ field: "startedAt" }], + }); + return rows.map((entry) => entry.data); + }); + + return { handleEvent, list, get, listToolCalls }; +}; + +/** Build an ExecutionObserver over a store instance — every engine event is + * forwarded to the store's buffered-batch writer. */ +export const makeExecutionHistoryObserver = ( + store: Pick, +): ExecutionObserver => ({ + handle: (event) => store.handleEvent(event), +}); diff --git a/packages/plugins/execution-history/tsconfig.json b/packages/plugins/execution-history/tsconfig.json new file mode 100644 index 0000000000..c495442f10 --- /dev/null +++ b/packages/plugins/execution-history/tsconfig.json @@ -0,0 +1,25 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "Bundler", + "strict": true, + "skipLibCheck": true, + "lib": ["ES2022", "DOM"], + "types": ["bun-types", "node"], + "noUnusedLocals": true, + "noImplicitOverride": true, + "jsx": "react-jsx", + "plugins": [ + { + "name": "@effect/language-service", + "ignoreEffectSuggestionsInTscExitCode": true, + "ignoreEffectWarningsInTscExitCode": true, + "diagnosticSeverity": { + "preferSchemaOverJson": "off" + } + } + ] + }, + "include": ["src/**/*.ts", "src/**/*.tsx"] +} diff --git a/packages/plugins/execution-history/tsup.config.ts b/packages/plugins/execution-history/tsup.config.ts new file mode 100644 index 0000000000..800469114b --- /dev/null +++ b/packages/plugins/execution-history/tsup.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: { + index: "src/sdk/index.ts", + }, + format: ["esm"], + dts: false, + sourcemap: true, + clean: true, + external: [/^@executor-js\//, /^effect/, /^@effect\//], +}); diff --git a/packages/plugins/execution-history/vitest.config.ts b/packages/plugins/execution-history/vitest.config.ts new file mode 100644 index 0000000000..5bfa2d586e --- /dev/null +++ b/packages/plugins/execution-history/vitest.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["src/**/*.test.ts"], + passWithNoTests: true, + }, +});