diff --git a/.changeset/fuzzy-turns-rollback.md b/.changeset/fuzzy-turns-rollback.md new file mode 100644 index 0000000000..a0e029e483 --- /dev/null +++ b/.changeset/fuzzy-turns-rollback.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Hooks can inspect a completed conversation turn with `beforeResponseRelease` and restore model history to an earlier index before terminal channel delivery. Restoring history suppresses the pending response while preserving earlier events and external side effects. diff --git a/.changeset/history-restoration-control.md b/.changeset/history-restoration-control.md new file mode 100644 index 0000000000..eccae2c934 --- /dev/null +++ b/.changeset/history-restoration-control.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Sessions can now restore model history to an earlier index with `restoreHistory({ to })`. The control is serialized with turns and retains only the selected history prefix without retracting prior events or external side effects. diff --git a/docs/guides/client/streaming.mdx b/docs/guides/client/streaming.mdx index 73923257a8..92cd8da9b6 100644 --- a/docs/guides/client/streaming.mdx +++ b/docs/guides/client/streaming.mdx @@ -38,6 +38,16 @@ const cleared = await session.clear(); console.log(cleared.status); ``` +Use `session.restoreHistory({ to })` to retain the exact model-history prefix before an index. The +control is serialized with turns and other session controls. It does not retract stream events, +messages already delivered through a channel, tool effects, or other external work. An invalid index +fails the session operation instead of selecting a nearby boundary. + +```ts +const restored = await session.restoreHistory({ to: historyIndex }); +console.log(restored.status); +``` + ## Aggregate a turn Use `result()` when you only need the final turn summary: diff --git a/docs/guides/hooks.md b/docs/guides/hooks.md index c169d92e37..e0ba5d2cc2 100644 --- a/docs/guides/hooks.md +++ b/docs/guides/hooks.md @@ -26,7 +26,43 @@ The slug is the path-relative basename. `agent/hooks/audit.ts` becomes `"audit"` `defineHook`, `HookDefinition`, and `HookContext` live on `eve/hooks`. -A hook file declares stream-event subscribers under the `events` map, keyed by event type, with `*` matching every event. Subscribe to any event in the runtime stream vocabulary documented in [Sessions, runs and streaming](../concepts/sessions-runs-and-streaming), including the lifecycle events `session.started`, `turn.completed`, `message.completed`, `action.partial`, and `action.result`. Handlers are observe-only. They cannot inject model context. To contribute runtime model messages, use `defineDynamic` and `defineInstructions` in `agent/instructions/`. +A hook file declares stream-event subscribers under the `events` map, keyed by event type, with `*` matching every event. Subscribe to any event in the runtime stream vocabulary documented in [Sessions, runs and streaming](../concepts/sessions-runs-and-streaming), including the lifecycle events `session.started`, `turn.completed`, `message.completed`, `action.partial`, and `action.result`. Stream-event handlers are observe-only. They cannot inject model context. To contribute runtime model messages, use `defineDynamic` and `defineInstructions` in `agent/instructions/`. + +## Gate terminal response release + +Use `beforeResponseRelease` when application policy must inspect a completed response before eve +releases its terminal completion to the channel: + +```ts title="agent/hooks/review.ts" +import { defineHook } from "eve/hooks"; + +export default defineHook({ + beforeResponseRelease(candidate) { + const questionIndex = candidate.history.messages.findLastIndex( + (message) => message.role === "user", + ); + if (questionIndex < 0) return; + const attempt = candidate.history.messages.slice(questionIndex); + + if (containsSensitiveData(attempt) && !containsRequiredApproval(attempt)) { + candidate.history.restoreTo(questionIndex); + } + }, +}); +``` + +The hook receives candidate model history, terminal output, and `turnId`. Calling +`candidate.history.restoreTo(index)` retains the exact history prefix before `index` and suppresses +the withheld terminal `message.completed` event. If several hooks request restoration, eve retains +the shortest requested prefix. If no hook requests restoration, eve keeps the candidate history and +releases the terminal event. + +This is a logical history-restoration boundary, not a private execution environment. Earlier events +have already run through channel handlers, the durable stream, memory, instrumentation, and ordinary +hooks. Model providers, tools, external systems, sandboxes, subagents, and background tasks may also +retain or continue acting on candidate content. The hook itself cannot durably pause for HITL; use a +tool or workflow for durable human input, then inspect its model-visible record before response +release. ## Scope side effects to a channel diff --git a/packages/eve/extension-contracts/compatibility/channel/v17.ts b/packages/eve/extension-contracts/compatibility/channel/v17.ts new file mode 100644 index 0000000000..67e58f0bbe --- /dev/null +++ b/packages/eve/extension-contracts/compatibility/channel/v17.ts @@ -0,0 +1,3 @@ +import { disableRoute } from "#public/channels/index.js"; + +export default disableRoute(); diff --git a/packages/eve/extension-contracts/compatibility/hook/v20.ts b/packages/eve/extension-contracts/compatibility/hook/v20.ts new file mode 100644 index 0000000000..c32db5c59d --- /dev/null +++ b/packages/eve/extension-contracts/compatibility/hook/v20.ts @@ -0,0 +1,9 @@ +import { defineHook } from "#public/hooks/index.js"; + +export default defineHook({ + events: { + "subagent.completed"(event, ctx) { + console.info(event.data.subagentName, ctx.session.id); + }, + }, +}); diff --git a/packages/eve/extension-contracts/compatibility/schedule/v9.ts b/packages/eve/extension-contracts/compatibility/schedule/v9.ts new file mode 100644 index 0000000000..58d9be2259 --- /dev/null +++ b/packages/eve/extension-contracts/compatibility/schedule/v9.ts @@ -0,0 +1,8 @@ +import { defineSchedule } from "#public/schedules/index.js"; + +export default defineSchedule({ + cron: "0 0 * * *", + run({ waitUntil }) { + waitUntil(Promise.resolve()); + }, +}); diff --git a/packages/eve/extension-contracts/reports/channel/v18.json b/packages/eve/extension-contracts/reports/channel/v18.json new file mode 100644 index 0000000000..9be3f783e0 --- /dev/null +++ b/packages/eve/extension-contracts/reports/channel/v18.json @@ -0,0 +1,21 @@ +{ + "kind": "eve-extension-capability-contract", + "capability": "channel", + "epoch": 18, + "sha256": "aa09c65ec5aa52fa7366a3ee56995d9d03e6612e4bb671438a648c5b035b474f", + "exports": [ + "DELETE", + "GET", + "HEAD", + "OPTIONS", + "PATCH", + "POST", + "PUT", + "WS", + "createWebSocketUpgradeServer", + "defineChannel", + "disableRoute", + "isChannel", + "isDisabledRouteSentinel" + ] +} diff --git a/packages/eve/extension-contracts/reports/hook/v21.json b/packages/eve/extension-contracts/reports/hook/v21.json new file mode 100644 index 0000000000..46a7338a55 --- /dev/null +++ b/packages/eve/extension-contracts/reports/hook/v21.json @@ -0,0 +1,7 @@ +{ + "kind": "eve-extension-capability-contract", + "capability": "hook", + "epoch": 21, + "sha256": "15d70241a9cb0f3d8e87b6309ce24e7c62b8bf620f065fb6a94bc95db1bf0a8f", + "exports": ["defineHook"] +} diff --git a/packages/eve/extension-contracts/reports/schedule/v10.json b/packages/eve/extension-contracts/reports/schedule/v10.json new file mode 100644 index 0000000000..5c7780afa4 --- /dev/null +++ b/packages/eve/extension-contracts/reports/schedule/v10.json @@ -0,0 +1,14 @@ +{ + "kind": "eve-extension-capability-contract", + "capability": "schedule", + "epoch": 10, + "sha256": "3a7852055fdd915739b6e08c2d09eba68be4bb76e23fab329c44f7348f0953f3", + "exports": [ + "ScheduleDefinition", + "ScheduleHandlerArgs", + "ScheduleRunHandler", + "ScheduleToFn", + "TypedReceiveTarget", + "defineSchedule" + ] +} diff --git a/packages/eve/src/channel/channel-address.ts b/packages/eve/src/channel/channel-address.ts index 4e9df5d576..970326a794 100644 --- a/packages/eve/src/channel/channel-address.ts +++ b/packages/eve/src/channel/channel-address.ts @@ -57,6 +57,7 @@ export interface ChannelAddress { cancel(options?: { readonly turnId?: string }): Promise; compact(): Promise; clear(): Promise; + restoreHistory(input: { readonly to: number }): Promise; reset(options?: { readonly reason?: string }): Promise; resolveSession(): Promise; } @@ -190,6 +191,12 @@ export function createChannelAddress(input: { continuationToken: namespacedToken, }); }, + async restoreHistory(restoreInput) { + return await input.runtime.dispatchContinuation({ + command: { kind: "restore-history", to: restoreInput.to }, + continuationToken: namespacedToken, + }); + }, async reset(options) { return await input.runtime.dispatchContinuation({ command: { kind: "reset", reason: options?.reason }, diff --git a/packages/eve/src/channel/channel-operations.ts b/packages/eve/src/channel/channel-operations.ts index ddf457037f..2f6a771559 100644 --- a/packages/eve/src/channel/channel-operations.ts +++ b/packages/eve/src/channel/channel-operations.ts @@ -67,6 +67,8 @@ export interface ChannelSource { compact(): Promise; /** Clears model-message history without creating a session. */ clear(): Promise; + /** Restores model-message history without creating a session. */ + restoreHistory(input: { readonly to: number }): Promise; /** Retires the current owner without creating a replacement. */ reset(options?: { readonly reason?: string }): Promise; } @@ -141,6 +143,9 @@ export function createChannelOperations(input: { async clear() { return await bound.clear(); }, + async restoreHistory(restoreInput) { + return await bound.restoreHistory(restoreInput); + }, async reset(options) { return await bound.reset(options); }, diff --git a/packages/eve/src/channel/cross-channel-receive.test.ts b/packages/eve/src/channel/cross-channel-receive.test.ts index 7bc21fb2c5..8efd8dda80 100644 --- a/packages/eve/src/channel/cross-channel-receive.test.ts +++ b/packages/eve/src/channel/cross-channel-receive.test.ts @@ -45,6 +45,9 @@ function makeSession(): Session { async clear() { return { status: "no_active_session" }; }, + async restoreHistory() { + return { status: "no_active_session" }; + }, async reset() { return { status: "no_active_session" }; }, diff --git a/packages/eve/src/channel/session.test.ts b/packages/eve/src/channel/session.test.ts index c98dae2e6c..86b08143ff 100644 --- a/packages/eve/src/channel/session.test.ts +++ b/packages/eve/src/channel/session.test.ts @@ -143,6 +143,7 @@ describe("fixed session operations", () => { await session.respond([{ optionId: "approve", requestId: "request_1" }], { auth: null }); await session.compact(); await session.clear(); + await session.restoreHistory({ to: 2 }); await session.reset({ reason: "fresh start" }); expect(runtime.dispatchSession).toHaveBeenNthCalledWith(1, { @@ -173,6 +174,10 @@ describe("fixed session operations", () => { sessionId: "sess_1", }); expect(runtime.dispatchSession).toHaveBeenNthCalledWith(5, { + command: { kind: "restore-history", to: 2 }, + sessionId: "sess_1", + }); + expect(runtime.dispatchSession).toHaveBeenNthCalledWith(6, { command: { kind: "reset", reason: "fresh start" }, sessionId: "sess_1", }); diff --git a/packages/eve/src/channel/session.ts b/packages/eve/src/channel/session.ts index ba955d39b9..e181595f9d 100644 --- a/packages/eve/src/channel/session.ts +++ b/packages/eve/src/channel/session.ts @@ -11,6 +11,7 @@ import type { ClearSessionResult, CompactSessionResult, ResetSessionResult, + RestoreHistorySessionResult, Runtime, SessionAuthContext, SessionCallback, @@ -54,6 +55,8 @@ export interface Session { compact(): Promise; /** Queues a context clear on this exact session ID. */ clear(): Promise; + /** Queues restoration of an observed model-history snapshot. */ + restoreHistory(input: { readonly to: number }): Promise; /** Terminally retires this exact session ID. */ reset(options?: { reason?: string }): Promise; getEventStream(options?: { startIndex?: number }): Promise>; @@ -162,6 +165,12 @@ export function createSession( async clear() { return await runtime.dispatchSession({ command: { kind: "clear" }, sessionId: id }); }, + async restoreHistory(input) { + return await runtime.dispatchSession({ + command: { kind: "restore-history", to: input.to }, + sessionId: id, + }); + }, async reset(options) { return await runtime.dispatchSession({ command: { kind: "reset", reason: options?.reason }, diff --git a/packages/eve/src/channel/types.ts b/packages/eve/src/channel/types.ts index cca4a3e8a4..5129ff6cb7 100644 --- a/packages/eve/src/channel/types.ts +++ b/packages/eve/src/channel/types.ts @@ -56,6 +56,9 @@ export type ClearSessionResult = | { readonly status: "accepted"; readonly sessionId: string } | { readonly status: "no_active_session" }; +/** Result of queueing model-history restoration for a session. */ +export type RestoreHistorySessionResult = ClearSessionResult; + // --------------------------------------------------------------------------- // Lineage // --------------------------------------------------------------------------- @@ -225,6 +228,10 @@ export type SessionCommand = } | { readonly kind: "compact" } | { readonly kind: "clear" } + | { + readonly kind: "restore-history"; + readonly to: number; + } | { readonly kind: "reset"; readonly reason?: string }; export type SessionSendCommandResult = @@ -236,6 +243,11 @@ export type ResetSessionResult = | { readonly status: "reset"; readonly previousSessionId: string } | { readonly status: "no_active_session" }; +export type SessionControlCommand = Extract< + SessionCommand, + { readonly kind: "clear" | "compact" | "restore-history" | "reset" } +>; + export type SessionCommandResult = TCommand extends { readonly kind: "send" } ? SessionSendCommandResult @@ -245,7 +257,9 @@ export type SessionCommandResult { readonly command: TCommand; @@ -302,6 +316,12 @@ export interface ClearSessionHookPayload { readonly kind: "clear"; } +/** Requests restoration of an observed model-history snapshot to an earlier index. */ +export interface RestoreHistorySessionHookPayload { + readonly kind: "restore-history"; + readonly to: number; +} + /** * Results resumed back into a parked parent workflow by the work it * dispatched: child-produced subagent results and authored workflow tool @@ -382,6 +402,7 @@ export type HookPayload = | CompactSessionHookPayload | DeliverHookPayload | RuntimeActionResultHookPayload + | RestoreHistorySessionHookPayload | SessionTimeoutHookPayload | SubagentAuthorizationEventHookPayload | SubagentInputRequestHookPayload; diff --git a/packages/eve/src/client/index.ts b/packages/eve/src/client/index.ts index 894fc63406..527dede773 100644 --- a/packages/eve/src/client/index.ts +++ b/packages/eve/src/client/index.ts @@ -56,6 +56,7 @@ export type { MessageResult, RespondTurnOptions, ResetResult, + RestoreHistoryResult, ResolvedStreamReconnectPolicy, SendTurnInput, SendTurnOptions, diff --git a/packages/eve/src/client/session-controls.ts b/packages/eve/src/client/session-controls.ts index 33ef31a7dd..21e7cb006c 100644 --- a/packages/eve/src/client/session-controls.ts +++ b/packages/eve/src/client/session-controls.ts @@ -5,17 +5,20 @@ import type { ClientRedirectPolicy, CompactResult, ResetResult, + RestoreHistoryResult, } from "#client/types.js"; import { createClientUrl } from "#client/url.js"; import { CancelTurnResponseSchema } from "#protocol/cancel-turn.js"; import { ClearResponseSchema } from "#protocol/clear-session.js"; import { CompactResponseSchema } from "#protocol/compact-session.js"; import { ResetResponseSchema } from "#protocol/reset-session.js"; +import { RestoreHistoryResponseSchema } from "#protocol/restore-history.js"; import { createEveSessionCancelRoutePath, createEveSessionClearRoutePath, createEveSessionCompactRoutePath, createEveSessionResetRoutePath, + createEveSessionRestoreHistoryRoutePath, } from "#protocol/routes.js"; interface SessionControlContext { @@ -95,6 +98,29 @@ export async function compactClientSession(input: { : { status: "no_active_session" }; } +export async function restoreClientSessionHistory(input: { + readonly context: SessionControlContext; + readonly sessionId: string; + readonly to: number; +}): Promise { + const { payload } = await postJson({ + body: { to: input.to }, + context: input.context, + operation: "Restore history", + path: createEveSessionRestoreHistoryRoutePath(input.sessionId), + }); + const result = RestoreHistoryResponseSchema.safeParse(payload); + if ( + !result.success || + (result.data.status === "accepted" && result.data.sessionId !== input.sessionId) + ) { + throw new Error("History restoration route returned an invalid response."); + } + return result.data.status === "accepted" + ? { sessionId: result.data.sessionId, status: "accepted" } + : { status: "no_active_session" }; +} + export async function resetClientSession(input: { readonly context: SessionControlContext; readonly options?: { readonly reason?: string; readonly signal?: AbortSignal }; diff --git a/packages/eve/src/client/session.test.ts b/packages/eve/src/client/session.test.ts index 5fefa5db93..dac97da996 100644 --- a/packages/eve/src/client/session.test.ts +++ b/packages/eve/src/client/session.test.ts @@ -440,6 +440,35 @@ describe("ClientSession", () => { expect(JSON.parse(requests[0]!.body ?? "{}")).toEqual({}); }); + it("queues history restoration without clearing the local session cursor", async () => { + const state = { sessionId: "session_1", streamIndex: 4 }; + const requests: Array<{ body?: string; method: string; url: string }> = []; + vi.spyOn(globalThis, "fetch").mockImplementation(async (request, init) => { + const url = + typeof request === "string" ? request : request instanceof URL ? request.href : request.url; + requests.push({ + body: typeof init?.body === "string" ? init.body : undefined, + method: init?.method ?? "GET", + url, + }); + return Response.json( + { ok: true, sessionId: "session_1", status: "accepted" }, + { status: 202 }, + ); + }); + const session = createSession(state); + + await expect(session.restoreHistory({ to: 2 })).resolves.toEqual({ + sessionId: "session_1", + status: "accepted", + }); + + expect(session.state).toEqual(state); + expect(new URL(requests[0]!.url).pathname).toBe("/eve/v1/session/session_1/restore-history"); + expect(requests[0]!.method).toBe("POST"); + expect(JSON.parse(requests[0]!.body ?? "{}")).toEqual({ to: 2 }); + }); + it("queues compaction without clearing the local session cursor", async () => { const state = { sessionId: "session_1", streamIndex: 4 }; const requests: Array<{ body?: string; method: string; url: string }> = []; diff --git a/packages/eve/src/client/session.ts b/packages/eve/src/client/session.ts index b67960ce64..613f8eb89f 100644 --- a/packages/eve/src/client/session.ts +++ b/packages/eve/src/client/session.ts @@ -9,6 +9,7 @@ import { clearClientSession, compactClientSession, resetClientSession, + restoreClientSessionHistory, } from "#client/session-controls.js"; import { serializeOutputSchema } from "#tools/schema.js"; import { createClientUrl } from "#client/url.js"; @@ -21,6 +22,7 @@ import type { ClientRedirectPolicy, RespondTurnOptions, ResetResult, + RestoreHistoryResult, SendTurnInput, SendTurnOptions, SendTurnPayload, @@ -150,6 +152,15 @@ export class ClientSession { return await compactClientSession({ context: this.#context, sessionId: this.#state.sessionId }); } + /** Queues restoration to an earlier model-history index. */ + async restoreHistory(input: { readonly to: number }): Promise { + return await restoreClientSessionHistory({ + context: this.#context, + sessionId: this.#state.sessionId, + to: input.to, + }); + } + /** Terminally retires this exact session ID. The handle remains pinned to it. */ async reset(options?: { readonly reason?: string; diff --git a/packages/eve/src/client/types.ts b/packages/eve/src/client/types.ts index df8ffb8e30..7ab7b994d9 100644 --- a/packages/eve/src/client/types.ts +++ b/packages/eve/src/client/types.ts @@ -6,6 +6,7 @@ import type { CancelTurnResult } from "#protocol/cancel-turn.js"; import type { ClearStatus } from "#protocol/clear-session.js"; import type { CompactStatus } from "#protocol/compact-session.js"; import type { ResetStatus } from "#protocol/reset-session.js"; +import type { RestoreHistoryStatus } from "#protocol/restore-history.js"; import type { TurnPolicy } from "#channel/types.js"; import type { InputRequest, InputResponse } from "#shared/input.js"; import type { JsonObject } from "#shared/json.js"; @@ -239,6 +240,16 @@ export type ClearResult = readonly status: Extract; }; +/** Result of requesting model-history restoration for a client session. */ +export type RestoreHistoryResult = + | { + readonly sessionId: string; + readonly status: Extract; + } + | { + readonly status: Extract; + }; + /** Result of requesting context compaction for a client session. */ export type CompactResult = | { diff --git a/packages/eve/src/compiler/extension-compatibility.ts b/packages/eve/src/compiler/extension-compatibility.ts index e9c0b50340..112185df4b 100644 --- a/packages/eve/src/compiler/extension-compatibility.ts +++ b/packages/eve/src/compiler/extension-compatibility.ts @@ -54,15 +54,15 @@ const EXTENSION_CAPABILITY_CONTRACTS = { }, }, channel: { - current: 17, - supported: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 14, 15, 16, 17], + current: 18, + supported: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 14, 15, 16, 17, 18], dropped: { 12: "Message and reasoning append events now expose deltas instead of cumulative snapshots.", }, }, schedule: { - current: 9, - supported: [1, 2, 3, 4, 6, 7, 8, 9], + current: 10, + supported: [1, 2, 3, 4, 6, 7, 8, 9, 10], dropped: { 5: "Message and reasoning append events now expose deltas instead of cumulative snapshots.", }, @@ -85,8 +85,8 @@ const EXTENSION_CAPABILITY_CONTRACTS = { }, }, hook: { - current: 20, - supported: [10, 11, 12, 13, 14, 15, 17, 18, 19, 20], + current: 21, + supported: [10, 11, 12, 13, 14, 15, 17, 18, 19, 20, 21], dropped: { 1: "Model identity moved from session.started runtime metadata to step.started call attribution.", 2: "Model identity moved from session.started runtime metadata to step.started call attribution.", diff --git a/packages/eve/src/compiler/normalize-hook.ts b/packages/eve/src/compiler/normalize-hook.ts index 07bb80fe85..43c8c0049b 100644 --- a/packages/eve/src/compiler/normalize-hook.ts +++ b/packages/eve/src/compiler/normalize-hook.ts @@ -21,6 +21,13 @@ export async function compileHookEntry( }), `Expected the hook export "${source.exportName ?? "default"}" from "${source.logicalPath}" to return an object.`, ); + const beforeResponseRelease = loaded.beforeResponseRelease; + if (beforeResponseRelease !== undefined) { + expectFunction( + beforeResponseRelease, + `Expected the hook export "${source.exportName ?? "default"}" from "${source.logicalPath}" to provide a function for beforeResponseRelease.`, + ); + } const events = loaded.events === undefined ? {} diff --git a/packages/eve/src/context/hook-lifecycle.integration.test.ts b/packages/eve/src/context/hook-lifecycle.integration.test.ts index cea998520e..ff942948df 100644 --- a/packages/eve/src/context/hook-lifecycle.integration.test.ts +++ b/packages/eve/src/context/hook-lifecycle.integration.test.ts @@ -6,7 +6,7 @@ import type { UnstampedMessageStreamEvent } from "#protocol/message.js"; import { stampTestEvent } from "#internal/testing/events.js"; import { mockSandbox } from "#internal/testing/mocks/mock-sandbox.js"; import { ContextContainer, contextStorage } from "./container.js"; -import { dispatchStreamEventHooks } from "./hook-lifecycle.js"; +import { dispatchBeforeResponseReleaseHooks, dispatchStreamEventHooks } from "./hook-lifecycle.js"; import { BundleKey, ChannelKey, @@ -51,6 +51,7 @@ function buildCtx(): ContextContainer { function hook(slug: string, hooks: Partial): ResolvedHookDefinition { return { + beforeResponseRelease: hooks.beforeResponseRelease, events: hooks.events ?? {}, exportName: undefined, logicalPath: `hooks/${slug}.ts`, @@ -60,6 +61,31 @@ function hook(slug: string, hooks: Partial): ResolvedHoo }; } +describe("dispatchBeforeResponseReleaseHooks", () => { + it("runs every pre-release hook in order", async () => { + const calls: string[] = []; + const registry = createRuntimeHookRegistry([ + hook("first", { beforeResponseRelease: async () => void calls.push("first") }), + hook("second", { beforeResponseRelease: async () => void calls.push("second") }), + ]); + const ctx = buildCtx(); + + await contextStorage.run(ctx, () => + dispatchBeforeResponseReleaseHooks({ + candidate: { + history: { messages: [], restoreTo: () => {} }, + output: "candidate", + turnId: "turn_0", + }, + ctx, + registry, + }), + ); + + expect(calls).toEqual(["first", "second"]); + }); +}); + describe("dispatchStreamEventHooks", () => { it("invokes typed then wildcard subscribers and propagates errors", async () => { const calls: string[] = []; diff --git a/packages/eve/src/context/hook-lifecycle.ts b/packages/eve/src/context/hook-lifecycle.ts index 8f99bdd5bb..c82dc3f03b 100644 --- a/packages/eve/src/context/hook-lifecycle.ts +++ b/packages/eve/src/context/hook-lifecycle.ts @@ -1,6 +1,6 @@ import { getAdapterKind } from "#channel/adapter.js"; import type { MessageStreamEvent } from "#protocol/message.js"; -import type { HookContext } from "#public/definitions/hook.js"; +import type { HookContext, ResponseReleaseCandidate } from "#public/definitions/hook.js"; import type { RuntimeHookRegistry } from "#runtime/hooks/registry.js"; import { buildCallbackContext } from "#context/build-callback-context.js"; import type { ContextContainer } from "./container.js"; @@ -34,6 +34,18 @@ export async function dispatchStreamEventHooks(input: { } } +/** Runs ordered pre-release hooks. */ +export async function dispatchBeforeResponseReleaseHooks(input: { + readonly candidate: ResponseReleaseCandidate; + readonly ctx: ContextContainer; + readonly registry: RuntimeHookRegistry; +}): Promise { + const hookCtx = buildHookContext(input.ctx); + for (const entry of input.registry.beforeResponseRelease) { + await entry.handler(input.candidate, hookCtx); + } +} + /** Builds the {@link HookContext} surfaced to one handler. */ function buildHookContext(ctx: ContextContainer): HookContext { const bundle = ctx.require(BundleKey); diff --git a/packages/eve/src/eve-channel/index.ts b/packages/eve/src/eve-channel/index.ts index e5b41ab913..ef4fc010a7 100644 --- a/packages/eve/src/eve-channel/index.ts +++ b/packages/eve/src/eve-channel/index.ts @@ -38,6 +38,7 @@ import { EVE_SESSION_COMPACT_ROUTE_PATTERN, EVE_SESSION_ROUTE_PATTERN, EVE_SESSION_RESET_ROUTE_PATTERN, + EVE_SESSION_RESTORE_HISTORY_ROUTE_PATTERN, EVE_SESSION_STREAM_ROUTE_PATTERN, EVE_SUBAGENT_STREAM_ROUTE_PATTERN, EVE_TASK_INPUT_ROUTE_PATTERN, @@ -48,6 +49,7 @@ import type { CancelTurnResponse } from "#protocol/cancel-turn.js"; import type { ClearResponse } from "#protocol/clear-session.js"; import type { CompactResponse } from "#protocol/compact-session.js"; import type { ResetResponse } from "#protocol/reset-session.js"; +import type { RestoreHistoryResponse } from "#protocol/restore-history.js"; import { parseTraceparent } from "#protocol/traceparent.js"; import { readForwardedAudienceBaggage } from "#protocol/baggage.js"; import { @@ -66,6 +68,7 @@ import { parseIncludeTailIndex, parseJsonRequest, parseResetBody, + parseRestoreHistoryBody, parseSessionControlBody, parseSessionMessageBody, parseStartIndex, @@ -469,6 +472,40 @@ export function eveChannel(input: EveChannelInput): EveChannel { ); }), + POST(EVE_SESSION_RESTORE_HISTORY_ROUTE_PATTERN, async (req, { attachSession, params }) => { + const authResult = await routeAuth(req, input.auth); + if (authResult instanceof Response) return authResult; + const sessionId = requireSessionId(params); + if (sessionId instanceof Response) return sessionId; + const body = await parseRestoreHistoryBody(req); + if (body instanceof Response) return body; + let result: Awaited>; + try { + result = await attachSession(sessionId).restoreHistory(body); + } catch (error) { + const errorId = logError(log, "session-history restoration request failed", error, { + sessionId, + }); + return Response.json( + { error: "Failed to restore the session history.", errorId, ok: false }, + { status: 500 }, + ); + } + return Response.json( + result.status === "accepted" + ? ({ + ok: true, + sessionId: result.sessionId, + status: "accepted", + } satisfies RestoreHistoryResponse) + : ({ ok: true, status: "no_active_session" } satisfies RestoreHistoryResponse), + { + headers: { "cache-control": "no-store" }, + status: result.status === "accepted" ? 202 : 200, + }, + ); + }), + POST(EVE_SESSION_RESET_ROUTE_PATTERN, async (req, { attachSession, params }) => { const authResult = await routeAuth(req, input.auth); if (authResult instanceof Response) return authResult; diff --git a/packages/eve/src/eve-channel/request.ts b/packages/eve/src/eve-channel/request.ts index 6a525949a7..a7e3f3d493 100644 --- a/packages/eve/src/eve-channel/request.ts +++ b/packages/eve/src/eve-channel/request.ts @@ -30,6 +30,7 @@ import { } from "#public/channels/upload-policy.js"; import { isInputResponse, type ValidatedInputResponse } from "#shared/input.js"; import { parseJsonObject, type JsonObject } from "#shared/json.js"; +import { RestoreHistoryRequestSchema } from "#protocol/restore-history.js"; import type { RunMode } from "#shared/run-mode.js"; interface ParsedCreateBody { @@ -256,6 +257,23 @@ export async function parseResetBody( return reason === undefined ? {} : { reason }; } +export async function parseRestoreHistoryBody( + req: Request, +): Promise<{ readonly to: number } | Response> { + const payload = await parseOptionalJsonRequest(req); + if (payload instanceof Response) return payload; + const tokenRejection = rejectSessionContinuationToken(payload); + if (tokenRejection !== null) return tokenRejection; + const parsed = RestoreHistoryRequestSchema.safeParse(payload); + if (!parsed.success) { + return Response.json( + { error: "Expected 'to' to be a nonnegative integer.", ok: false }, + { status: 400 }, + ); + } + return parsed.data; +} + export async function parseSessionControlBody( req: Request, ): Promise | Response> { diff --git a/packages/eve/src/execution/inline-turn.ts b/packages/eve/src/execution/inline-turn.ts index ecdbd28eb3..f2dd64aaf4 100644 --- a/packages/eve/src/execution/inline-turn.ts +++ b/packages/eve/src/execution/inline-turn.ts @@ -1,4 +1,6 @@ import type { DeliverHookPayload, HookPayload, SessionCapabilities } from "#channel/types.js"; +import type { BufferedSessionControl } from "#execution/parked-delivery-wait.js"; + import { readAcceptedDeploymentId } from "#execution/accepted-delivery-deployment.js"; import { cancelAllIndexedSessionTasksStep } from "#execution/cancel-indexed-session-tasks-step.js"; import type { @@ -32,7 +34,7 @@ export type InlineTurnOutcome = /** Runs same-deployment steps directly until they need the shared turn runner. */ export async function runInlineTurn(input: { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds?: Set; readonly capabilities?: SessionCapabilities; readonly commandInbox: SessionCommandInbox; @@ -159,7 +161,7 @@ function continueOutcome( class InlineTurnControl { private readonly bufferedDeliveries: DeliverHookPayload[]; - private readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + private readonly bufferedSessionControls: BufferedSessionControl[]; private readonly cancelledTaskIds: Set; private readonly commandInbox: SessionCommandInbox; private readonly controller = new AbortController(); @@ -170,7 +172,7 @@ class InlineTurnControl { constructor(input: { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds?: Set; readonly commandInbox: SessionCommandInbox; readonly expectedTurnId: string; @@ -242,16 +244,20 @@ class InlineTurnControl { if (command.turnPolicy === "steer" && deliveryHasMessage(command)) this.abort({}); return; } - if (command.kind === "clear" || command.kind === "compact") { - this.bufferedSessionControls.push(command.kind); + if ( + command.kind === "clear" || + command.kind === "compact" || + command.kind === "restore-history" + ) { + this.bufferedSessionControls.push(command); return; } if (command.kind === "session-timeout") { - this.bufferedSessionControls.push("expired"); + this.bufferedSessionControls.push({ kind: "expired" }); return; } if (command.kind === "reset") { - this.bufferedSessionControls.push("reset"); + this.bufferedSessionControls.push(command); this.abort({}); return; } diff --git a/packages/eve/src/execution/node-step.ts b/packages/eve/src/execution/node-step.ts index 3b4a627037..ca368b41c9 100644 --- a/packages/eve/src/execution/node-step.ts +++ b/packages/eve/src/execution/node-step.ts @@ -62,12 +62,17 @@ export interface CreateExecutionNodeStepInput { readonly clearOnly?: boolean; /** Runs only a forced context compaction and returns to the parked session. */ readonly compactOnly?: boolean; + /** Restores model history and returns to the parked session. */ + readonly restoreHistoryTo?: number; /** * Runtime constructor used by the subagent tool executor to start * delegated child runs on the same workflow runtime as the parent. */ readonly createRuntime: CreateRuntime; readonly handleEvent?: HandleEventFn; + readonly beforeResponseRelease?: Parameters< + typeof createToolLoopHarness + >[0]["beforeResponseRelease"]; readonly historyProjector?: HistoryViewProjector; readonly historyView?: PreparedHistoryView; readonly instrumentation: ExecutionInstrumentation | undefined; @@ -102,8 +107,10 @@ export function createExecutionNodeStep(input: CreateExecutionNodeStepInput): St capabilities: input.capabilities, clearOnly: input.clearOnly, compactOnly: input.compactOnly, + restoreHistoryTo: input.restoreHistoryTo, workflow: input.node.agent.workflowTool !== undefined, workflowMaxSubagents: input.workflowMaxSubagents, + beforeResponseRelease: input.beforeResponseRelease, handleEvent: input.handleEvent, historyProjector: input.historyProjector, historyView: input.historyView, diff --git a/packages/eve/src/execution/parked-delivery-wait.ts b/packages/eve/src/execution/parked-delivery-wait.ts index 6389215fb3..7c67f07a4e 100644 --- a/packages/eve/src/execution/parked-delivery-wait.ts +++ b/packages/eve/src/execution/parked-delivery-wait.ts @@ -1,4 +1,4 @@ -import type { DeliverHookPayload, DeliverPayload } from "#channel/types.js"; +import type { DeliverHookPayload, DeliverPayload, SessionControlCommand } from "#channel/types.js"; import { cancelAllIndexedSessionTasksStep } from "#execution/cancel-indexed-session-tasks-step.js"; import { routeDeliverToChildren } from "#execution/route-child-delivery.js"; import type { SessionCommandInbox } from "#execution/session-command-inbox.js"; @@ -11,11 +11,10 @@ import { } from "#execution/wire/session-inbox-wire.js"; import { coalesceDeliveries } from "#harness/messages.js"; +export type BufferedSessionControl = SessionControlCommand | { readonly kind: "expired" }; + type NextSessionAction = - | { readonly kind: "clear" } - | { readonly kind: "compact" } - | { readonly kind: "expired" } - | { readonly kind: "reset" } + | BufferedSessionControl | AuthorizationCallbackInstruction | { readonly delivery: DeliverHookPayload | null; @@ -32,10 +31,7 @@ export interface AuthorizationCallbackInstruction { /** What the parked driver should do with the next session activity. */ export type NextTurnInstruction = - | { readonly kind: "clear" } - | { readonly kind: "compact" } - | { readonly kind: "expired" } - | { readonly kind: "reset" } + | BufferedSessionControl | { readonly kind: "closed" } | { readonly kind: "cancel-turn" } | AuthorizationCallbackInstruction @@ -63,7 +59,7 @@ export type NextTurnInstruction = export async function nextTurnDelivery(input: { readonly awaitAuthorizationCallbacks?: boolean; readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds?: Set; readonly commandInbox: SessionCommandInbox; readonly deferDeliveries?: boolean; @@ -85,7 +81,7 @@ export async function nextTurnDelivery(input: { async function awaitNextTurnDelivery(input: { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds?: Set; readonly commandInbox: SessionCommandInbox; readonly deferDeliveries?: boolean; @@ -110,9 +106,7 @@ async function awaitNextTurnDelivery(input: { return nextAction; } - if (nextAction.kind !== "delivery") { - return { kind: nextAction.kind }; - } + if (nextAction.kind !== "delivery") return nextAction; const deliver = nextAction.delivery; if (deliver === null) { @@ -142,7 +136,7 @@ async function awaitNextTurnDelivery(input: { async function waitForNextSessionAction(input: { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds: Set; readonly commandInbox: SessionCommandInbox; readonly deferDeliveries?: boolean; @@ -150,9 +144,7 @@ async function waitForNextSessionAction(input: { readonly stateCursor: SessionStateCursor; }): Promise { const pendingSessionControl = input.bufferedSessionControls.shift(); - if (pendingSessionControl !== undefined) { - return { kind: pendingSessionControl }; - } + if (pendingSessionControl !== undefined) return pendingSessionControl; while ( input.bufferedDeliveries[0] !== undefined && @@ -212,8 +204,13 @@ async function waitForNextSessionAction(input: { return { kind: "expired" }; } - if (decoded.kind === "clear" || decoded.kind === "compact" || decoded.kind === "reset") { - return { kind: decoded.kind }; + if ( + decoded.kind === "clear" || + decoded.kind === "compact" || + decoded.kind === "reset" || + decoded.kind === "restore-history" + ) { + return decoded; } if (decoded.kind === "cancel") { diff --git a/packages/eve/src/execution/response-release-event-gate.test.ts b/packages/eve/src/execution/response-release-event-gate.test.ts new file mode 100644 index 0000000000..56b7493923 --- /dev/null +++ b/packages/eve/src/execution/response-release-event-gate.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it, vi } from "vitest"; + +import { ContextContainer } from "#context/container.js"; +import { ResponseReleaseEventGate } from "#execution/response-release-event-gate.js"; +import type { RuntimeHookRegistry } from "#runtime/hooks/registry.js"; + +const { dispatchBeforeResponseReleaseHooks } = vi.hoisted(() => ({ + dispatchBeforeResponseReleaseHooks: vi.fn(), +})); + +vi.mock("#context/hook-lifecycle.js", () => ({ dispatchBeforeResponseReleaseHooks })); + +const terminal = { + data: { + finishReason: "stop" as const, + message: "candidate", + sequence: 0, + stepIndex: 0, + turnId: "turn_0", + }, + type: "message.completed" as const, +}; + +const registry: RuntimeHookRegistry = { + beforeResponseRelease: [{ handler: vi.fn(), slug: "review" }], + streamEventsByType: new Map(), + streamEventsWildcard: [], +}; + +describe("ResponseReleaseEventGate", () => { + it("withholds then releases a terminal completion when history is retained", async () => { + const gate = new ResponseReleaseEventGate(new ContextContainer(), registry); + const release = vi.fn().mockResolvedValue(undefined); + + expect(gate.intercept(terminal)).toBe(true); + await expect( + gate.beforeRelease(release)!({ history: [], output: "candidate", turnId: "turn_0" }), + ).resolves.toBeUndefined(); + expect(release).toHaveBeenCalledWith(terminal); + }); + + it("does not intercept task-mode terminal completions", () => { + const gate = new ResponseReleaseEventGate(new ContextContainer(), registry, false); + + expect(gate.intercept(terminal)).toBe(false); + expect(gate.beforeRelease(vi.fn())).toBeUndefined(); + }); + + it("does not intercept a response that parks on tool calls", () => { + const gate = new ResponseReleaseEventGate(new ContextContainer(), registry); + + expect( + gate.intercept({ + ...terminal, + data: { ...terminal.data, finishReason: "tool-calls" }, + }), + ).toBe(false); + }); + + it("drops a terminal completion when a hook requests history restoration", async () => { + dispatchBeforeResponseReleaseHooks.mockImplementationOnce(async ({ candidate }) => { + candidate.history.restoreTo(1); + }); + const gate = new ResponseReleaseEventGate(new ContextContainer(), registry); + const release = vi.fn().mockResolvedValue(undefined); + + expect(gate.intercept(terminal)).toBe(true); + await expect( + gate.beforeRelease(release)!({ + history: [ + { content: "keep", role: "user" }, + { content: "remove", role: "assistant" }, + ], + output: "candidate", + turnId: "turn_0", + }), + ).resolves.toBe(1); + expect(release).not.toHaveBeenCalled(); + }); + + it("uses the earliest restoration requested by several hooks", async () => { + dispatchBeforeResponseReleaseHooks.mockImplementationOnce(async ({ candidate }) => { + candidate.history.restoreTo(2); + candidate.history.restoreTo(1); + }); + const gate = new ResponseReleaseEventGate(new ContextContainer(), registry); + + await expect( + gate.beforeRelease(vi.fn())!({ + history: [ + { content: "keep", role: "user" }, + { content: "remove", role: "assistant" }, + { content: "remove too", role: "user" }, + ], + output: "candidate", + turnId: "turn_0", + }), + ).resolves.toBe(1); + }); +}); diff --git a/packages/eve/src/execution/response-release-event-gate.ts b/packages/eve/src/execution/response-release-event-gate.ts new file mode 100644 index 0000000000..9522c86003 --- /dev/null +++ b/packages/eve/src/execution/response-release-event-gate.ts @@ -0,0 +1,78 @@ +import type { ContextContainer } from "#context/container.js"; +import { dispatchBeforeResponseReleaseHooks } from "#context/hook-lifecycle.js"; +import { validateHistoryRestoreIndex } from "#harness/history-restoration.js"; +import type { ToolLoopHarnessConfig } from "#harness/types.js"; +import type { UnstampedMessageStreamEvent } from "#protocol/message.js"; +import type { RuntimeHookRegistry } from "#runtime/hooks/registry.js"; + +/** Holds terminal content while authored hooks inspect the settling turn. */ +export class ResponseReleaseEventGate { + private readonly ctx: ContextContainer; + private readonly registry: RuntimeHookRegistry; + private readonly supported: boolean; + private releasing = false; + private terminalEvent: UnstampedMessageStreamEvent | undefined; + + constructor(ctx: ContextContainer, registry: RuntimeHookRegistry, supported = true) { + this.ctx = ctx; + this.registry = registry; + this.supported = supported; + } + + get enabled(): boolean { + return this.supported && this.registry.beforeResponseRelease.length > 0; + } + + /** Returns true when the terminal event was withheld from ordinary delivery. */ + intercept(event: UnstampedMessageStreamEvent): boolean { + if ( + this.releasing || + !this.enabled || + event.type !== "message.completed" || + event.data.finishReason === "tool-calls" + ) { + return false; + } + this.terminalEvent = event; + return true; + } + + beforeRelease( + release: (event: UnstampedMessageStreamEvent) => Promise, + ): NonNullable | undefined { + if (!this.enabled) return undefined; + return async (candidate) => { + let restoreHistoryTo: number | undefined; + await dispatchBeforeResponseReleaseHooks({ + candidate: { + history: { + messages: candidate.history, + restoreTo(index) { + validateHistoryRestoreIndex(candidate.history.length, index); + restoreHistoryTo = Math.min(restoreHistoryTo ?? index, index); + }, + }, + output: candidate.output, + turnId: candidate.turnId, + }, + ctx: this.ctx, + registry: this.registry, + }); + if (restoreHistoryTo !== undefined) { + this.terminalEvent = undefined; + return restoreHistoryTo; + } + if (this.terminalEvent !== undefined) { + const terminalEvent = this.terminalEvent; + this.terminalEvent = undefined; + this.releasing = true; + try { + await release(terminalEvent); + } finally { + this.releasing = false; + } + } + return undefined; + }; + } +} diff --git a/packages/eve/src/execution/turn-control-receiver.test.ts b/packages/eve/src/execution/turn-control-receiver.test.ts index 1fa2788817..b75e47e02d 100644 --- a/packages/eve/src/execution/turn-control-receiver.test.ts +++ b/packages/eve/src/execution/turn-control-receiver.test.ts @@ -9,6 +9,7 @@ import { forwardTurnDeliveryStep } from "#execution/forward-turn-delivery-step.j import { reportDroppedWirePayloadStep } from "#execution/report-dropped-wire-payload-step.js"; import type { SessionCommandInbox, SessionInboxPayload } from "#execution/session-command-inbox.js"; import type { TurnControlPayload } from "#execution/turn-control-protocol.js"; +import type { BufferedSessionControl } from "#execution/parked-delivery-wait.js"; import { TurnControlReceiver } from "#execution/turn-control-receiver.js"; const createHookMock = vi.fn(); @@ -128,7 +129,7 @@ describe("TurnControlReceiver", () => { it("buffers current and legacy sends that arrive during a turn", async () => { installControlHook([parkResult()], true); const bufferedDeliveries: DeliverHookPayload[] = []; - const bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset"> = []; + const bufferedSessionControls: BufferedSessionControl[] = []; const action = await runReceiver(bufferedDeliveries, { bufferedSessionControls, @@ -137,6 +138,7 @@ describe("TurnControlReceiver", () => { { kind: "deliver", payloads: [{ message: "legacy follow up" }] }, { kind: "clear" }, { kind: "compact" }, + { kind: "restore-history", to: 2 }, { kind: "session-timeout" }, ]), }); @@ -154,7 +156,12 @@ describe("TurnControlReceiver", () => { }, { kind: "deliver", payloads: [{ message: "legacy follow up" }] }, ]); - expect(bufferedSessionControls).toEqual(["clear", "compact", "expired"]); + expect(bufferedSessionControls).toEqual([ + { kind: "clear" }, + { kind: "compact" }, + { kind: "restore-history", to: 2 }, + { kind: "expired" }, + ]); }); it("consumes a replayed task delivery only once", async () => { @@ -275,7 +282,7 @@ describe("TurnControlReceiver", () => { it("forwards cancel and reset through the active turn's private hook", async () => { installControlHook([parkResult()], true); - const bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset"> = []; + const bufferedSessionControls: BufferedSessionControl[] = []; const action = await runReceiver([], { bufferedSessionControls, @@ -294,14 +301,14 @@ describe("TurnControlReceiver", () => { payload: {}, token: "turn-control:cancel", }); - expect(bufferedSessionControls).toEqual(["reset"]); + expect(bufferedSessionControls).toEqual([{ kind: "reset", reason: "Start over" }]); }); }); function runReceiver( bufferedDeliveries: DeliverHookPayload[], options: { - readonly bufferedSessionControls?: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls?: BufferedSessionControl[]; readonly commandInbox?: SessionCommandInbox; readonly seenTaskDeliveries?: Set; } = {}, diff --git a/packages/eve/src/execution/turn-control-receiver.ts b/packages/eve/src/execution/turn-control-receiver.ts index 653158781f..d5e60ab3cc 100644 --- a/packages/eve/src/execution/turn-control-receiver.ts +++ b/packages/eve/src/execution/turn-control-receiver.ts @@ -7,6 +7,7 @@ import type { TurnControlPayload } from "#execution/turn-control-protocol.js"; import { forwardTurnDeliveryStep } from "#execution/forward-turn-delivery-step.js"; import { closeHookIterator, disposeHook } from "#execution/hook-ownership.js"; import type { NextDriverAction } from "#execution/next-driver-action.js"; +import type { BufferedSessionControl } from "#execution/parked-delivery-wait.js"; import type { SessionCommandInbox } from "#execution/session-command-inbox.js"; import type { SessionStateCursor } from "#execution/session-state-cursor.js"; import { turnCancellationHookToken } from "#execution/turn-cancellation-token.js"; @@ -25,7 +26,7 @@ export type TurnDriverAction = NextDriverAction; /** Owns one turn's driver-side control hook and public-delivery relay state. */ export class TurnControlReceiver { private readonly bufferedDeliveries: DeliverHookPayload[]; - private readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + private readonly bufferedSessionControls: BufferedSessionControl[]; private readonly commandInbox: SessionCommandInbox; private readonly control: Hook; private readonly controlIterator: AsyncIterator; @@ -37,7 +38,7 @@ export class TurnControlReceiver { constructor(input: { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly cancelledTaskIds?: Set; readonly commandInbox: SessionCommandInbox; readonly expectedTurnId: string; @@ -96,12 +97,16 @@ export class TurnControlReceiver { await this.bufferDelivery(command); return undefined; } - if (command.kind === "clear" || command.kind === "compact") { - this.bufferedSessionControls.push(command.kind); + if ( + command.kind === "clear" || + command.kind === "compact" || + command.kind === "restore-history" + ) { + this.bufferedSessionControls.push(command); return undefined; } if (command.kind === "session-timeout") { - this.bufferedSessionControls.push("expired"); + this.bufferedSessionControls.push({ kind: "expired" }); return undefined; } if (command.kind === "cancel") { @@ -131,7 +136,7 @@ export class TurnControlReceiver { payload: {}, token: turnCancellationHookToken(this.control.token), }); - this.bufferedSessionControls.push("reset"); + this.bufferedSessionControls.push(command); return undefined; } return unsupportedSessionCommand(command); diff --git a/packages/eve/src/execution/turn-dispatch.ts b/packages/eve/src/execution/turn-dispatch.ts index e757e56f1f..918ea8fcab 100644 --- a/packages/eve/src/execution/turn-dispatch.ts +++ b/packages/eve/src/execution/turn-dispatch.ts @@ -8,6 +8,7 @@ import { type TurnWorkflowDispatchInput, } from "#execution/durable-session-migrations/turn-workflow.js"; import { runInlineTurn } from "#execution/inline-turn.js"; +import type { BufferedSessionControl } from "#execution/parked-delivery-wait.js"; import type { SessionCommandInbox } from "#execution/session-command-inbox.js"; import { SessionStateCursor } from "#execution/session-state-cursor.js"; import type { TurnCancelPayload } from "#execution/turn-cancellation-token.js"; @@ -33,7 +34,7 @@ export interface DispatchedTurn { /** Dispatches one turn and services its private-inbox control protocol until it terminates. */ interface TurnDispatchInput { readonly bufferedDeliveries: DeliverHookPayload[]; - readonly bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset">; + readonly bufferedSessionControls: BufferedSessionControl[]; readonly capabilities?: SessionCapabilities; readonly cancelledTaskIds?: Set; readonly controlToken: string; diff --git a/packages/eve/src/execution/wire/session-inbox-contract.ts b/packages/eve/src/execution/wire/session-inbox-contract.ts index b50fffbb4a..9116df6667 100644 --- a/packages/eve/src/execution/wire/session-inbox-contract.ts +++ b/packages/eve/src/execution/wire/session-inbox-contract.ts @@ -1,5 +1,5 @@ /** Every explicit session-inbox wire version still supported by producers. */ -export const SESSION_INBOX_WIRE_VERSIONS = [1, 2, 3, 4, 5, 6] as const; +export const SESSION_INBOX_WIRE_VERSIONS = [1, 2, 3, 4, 5, 6, 7] as const; export type SessionInboxWireVersion = (typeof SESSION_INBOX_WIRE_VERSIONS)[number]; diff --git a/packages/eve/src/execution/wire/session-inbox-encoder.ts b/packages/eve/src/execution/wire/session-inbox-encoder.ts index b3fa727c8b..8b483f3b7b 100644 --- a/packages/eve/src/execution/wire/session-inbox-encoder.ts +++ b/packages/eve/src/execution/wire/session-inbox-encoder.ts @@ -34,11 +34,15 @@ import { encodeSessionCommandV6, type SessionInboxWireV6, } from "#execution/wire/session-inbox-wire.v6.js"; +import { + encodeSessionCommandV7, + type SessionInboxWireV7, +} from "#execution/wire/session-inbox-wire.v7.js"; type SessionInboxCommand = DeliverHookPayload | SessionCommand | SessionTimeoutHookPayload; /** Current wire type consumed after migration. */ -export type SessionInboxWire = SessionInboxWireV6; +export type SessionInboxWire = SessionInboxWireV7; type LegacySessionInboxWireTarget = Extract; type VersionedSessionInboxEncoder = (command: SessionInboxCommand) => unknown; @@ -55,6 +59,7 @@ const versionedEncoders = { 5: (command: SessionInboxCommand) => encodeSessionCommandV5(withoutOwnedTaskCancellation(command)), 6: encodeSessionCommandV6, + 7: encodeSessionCommandV7, } satisfies Record; /** Encodes a command for the selected session-inbox consumer. */ @@ -64,6 +69,7 @@ function encode(command: SessionInboxCommand, target: { readonly version: 3 }): function encode(command: SessionInboxCommand, target: { readonly version: 4 }): SessionInboxWireV4; function encode(command: SessionInboxCommand, target: { readonly version: 5 }): SessionInboxWireV5; function encode(command: SessionInboxCommand, target: { readonly version: 6 }): SessionInboxWireV6; +function encode(command: SessionInboxCommand, target: { readonly version: 7 }): SessionInboxWireV7; function encode( command: SessionInboxCommand, target: { readonly version: SessionInboxWireVersion }, @@ -82,6 +88,7 @@ function encode( | SessionInboxWireV4 | SessionInboxWireV5 | SessionInboxWireV6 + | SessionInboxWireV7 | Record; function encode( command: SessionInboxCommand, @@ -93,7 +100,13 @@ function encode( | SessionInboxWireV4 | SessionInboxWireV5 | SessionInboxWireV6 + | SessionInboxWireV7 | Record { + if (command.kind === "restore-history" && target.version < 7) { + throw new SessionInboxWireError( + `Cannot encode history restoration for wire version ${target.version}.`, + ); + } if (command.kind === "cancel" && command.tasks === true && target.version < 6) { throw new SessionInboxWireError( `Cannot encode session-owned task cancellation for wire version ${target.version}.`, @@ -139,7 +152,8 @@ function encode( | SessionInboxWireV3 | SessionInboxWireV4 | SessionInboxWireV5 - | SessionInboxWireV6; + | SessionInboxWireV6 + | SessionInboxWireV7; } throw new SessionInboxWireError( `Cannot encode session inbox payload for unknown wire version ${JSON.stringify((target as { version?: unknown }).version)}.`, diff --git a/packages/eve/src/execution/wire/session-inbox-wire.ts b/packages/eve/src/execution/wire/session-inbox-wire.ts index 902a3d80e3..eb013f1c3c 100644 --- a/packages/eve/src/execution/wire/session-inbox-wire.ts +++ b/packages/eve/src/execution/wire/session-inbox-wire.ts @@ -19,6 +19,7 @@ import { sessionInboxWireV1Migration } from "#execution/wire/session-inbox-wire. import { sessionInboxWireV2Migration } from "#execution/wire/session-inbox-wire.v3.migration.js"; import { sessionInboxWireV3Migration } from "#execution/wire/session-inbox-wire.v4.migration.js"; import { sessionInboxWireV4Migration } from "#execution/wire/session-inbox-wire.v5.migration.js"; +import { sessionInboxWireV6Migration } from "#execution/wire/session-inbox-wire.v7.migration.js"; import { isObject } from "#shared/guards.js"; /** @@ -37,7 +38,10 @@ import { isObject } from "#shared/guards.js"; export type DecodedSessionInbox = | DeliverHookPayload | SessionTimeoutHookPayload - | Extract; + | Extract< + SessionCommand, + { readonly kind: "cancel" | "clear" | "compact" | "reset" | "restore-history" } + >; export { SessionInboxWireError } from "#execution/wire/session-inbox-contract.js"; @@ -60,6 +64,7 @@ const sessionInboxMigrations: readonly VersionMigration[] = [ sessionInboxWireV3Migration, sessionInboxWireV4Migration, sessionInboxWireV5Migration, + sessionInboxWireV6Migration, ]; /** @@ -160,6 +165,8 @@ function normalizeWire(wire: SessionInboxWire): DecodedSessionInbox { return { kind: "compact" }; case "reset": return { kind: "reset", reason: wire.reason }; + case "restore-history": + return { kind: "restore-history", to: wire.to }; case "cancel": return { kind: "cancel", taskId: wire.taskId, tasks: wire.tasks, turnId: wire.turnId }; default: diff --git a/packages/eve/src/execution/wire/session-inbox-wire.v7.migration.ts b/packages/eve/src/execution/wire/session-inbox-wire.v7.migration.ts new file mode 100644 index 0000000000..ad6a42f83a --- /dev/null +++ b/packages/eve/src/execution/wire/session-inbox-wire.v7.migration.ts @@ -0,0 +1,12 @@ +import type { VersionMigration } from "#execution/durable-session-migrations/chain.js"; +import { isObject } from "#shared/guards.js"; + +/** Advances version-6 payloads to the model-history restoration schema. */ +export const sessionInboxWireV6Migration: VersionMigration = { + from: 6, + migrate(prior) { + if (!isObject(prior)) throw new Error("session inbox wire v6 value is not an object."); + return { ...prior, version: 7 }; + }, + to: 7, +}; diff --git a/packages/eve/src/execution/wire/session-inbox-wire.v7.test.ts b/packages/eve/src/execution/wire/session-inbox-wire.v7.test.ts new file mode 100644 index 0000000000..0bfdf9194e --- /dev/null +++ b/packages/eve/src/execution/wire/session-inbox-wire.v7.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; + +import { sessionInboxWire } from "#execution/wire/session-inbox-encoder.js"; +import { SessionInboxWireError } from "#execution/wire/session-inbox-contract.js"; +import { sessionInboxWire as sessionInboxWireDecoder } from "#execution/wire/session-inbox-wire.js"; +import { sessionInboxWireV7Schema } from "#execution/wire/session-inbox-wire.v7.js"; + +describe("session inbox wire v7", () => { + it("round-trips history restoration", () => { + const wire = sessionInboxWire.encode({ kind: "restore-history", to: 3 }, { version: 7 }); + + expect(wire).toEqual({ kind: "restore-history", to: 3, version: 7 }); + expect(sessionInboxWireDecoder.decode(wire)).toEqual({ kind: "restore-history", to: 3 }); + }); + + it("keeps restoration strict", () => { + expect( + sessionInboxWireV7Schema.safeParse({ kind: "restore-history", to: -1, version: 7 }).success, + ).toBe(false); + expect( + sessionInboxWireV7Schema.safeParse({ + extra: true, + kind: "restore-history", + to: 1, + version: 7, + }).success, + ).toBe(false); + }); + + it.each([1, 2, 3, 4, 5, 6] as const)( + "fails closed for history restoration on v%i consumers", + (version) => { + expect(() => + sessionInboxWire.encode({ kind: "restore-history", to: 1 }, { version }), + ).toThrowError(SessionInboxWireError); + }, + ); +}); diff --git a/packages/eve/src/execution/wire/session-inbox-wire.v7.ts b/packages/eve/src/execution/wire/session-inbox-wire.v7.ts new file mode 100644 index 0000000000..9c13f6396f --- /dev/null +++ b/packages/eve/src/execution/wire/session-inbox-wire.v7.ts @@ -0,0 +1,54 @@ +import { z } from "#compiled/zod/index.js"; + +import type { + DeliverHookPayload, + SessionCommand, + SessionTimeoutHookPayload, +} from "#channel/types.js"; +import { SessionInboxWireError } from "#execution/wire/session-inbox-contract.js"; +import { + encodeSessionCommandV6, + sessionInboxWireV6Schema, +} from "#execution/wire/session-inbox-wire.v6.js"; +import { formatValidationError } from "#runtime/validation.js"; + +const VERSION = 7; +const version = z.literal(VERSION); + +/** Version 7 adds model-history restoration. */ +const v6 = sessionInboxWireV6Schema.options; + +export const sessionInboxWireV7Schema = z.discriminatedUnion("kind", [ + v6[0].extend({ version }), + v6[1].extend({ version }), + v6[2].extend({ version }), + v6[3].extend({ version }), + v6[4].extend({ version }), + v6[5].extend({ version }), + z + .object({ + kind: z.literal("restore-history"), + to: z.number().int().nonnegative(), + version, + }) + .strict(), +]); + +export type SessionInboxWireV7 = z.infer; + +/** Builds and validates one complete version-7 wire value. */ +export function encodeSessionCommandV7( + command: DeliverHookPayload | SessionCommand | SessionTimeoutHookPayload, +): SessionInboxWireV7 { + const value = + command.kind === "restore-history" + ? { ...command, version: VERSION } + : { ...encodeSessionCommandV6(command), version: VERSION }; + const parsed = sessionInboxWireV7Schema.safeParse(value); + if (!parsed.success) { + throw new SessionInboxWireError( + `Produced a session inbox payload that does not match wire version ${VERSION}: ${formatValidationError(parsed.error)}`, + ); + } + return parsed.data; +} diff --git a/packages/eve/src/execution/workflow-entry.ts b/packages/eve/src/execution/workflow-entry.ts index d3cfb0bd37..817dbedfee 100644 --- a/packages/eve/src/execution/workflow-entry.ts +++ b/packages/eve/src/execution/workflow-entry.ts @@ -1,3 +1,5 @@ +import type { BufferedSessionControl } from "#execution/parked-delivery-wait.js"; + import { getWorkflowMetadata, getWritable } from "#compiled/@workflow/core/index.js"; import type { @@ -473,7 +475,7 @@ async function runDriverLoop(input: { `${input.sessionState.sessionId}:turn-control:${String(turnDispatchIndex++)}`; const bufferedDeliveries: DeliverHookPayload[] = []; - const bufferedSessionControls: Array<"clear" | "compact" | "expired" | "reset"> = []; + const bufferedSessionControls: BufferedSessionControl[] = []; const cancelledTaskIds = new Set(); const seenTaskDeliveries = new Set(); const stateCursor = new SessionStateCursor({ @@ -644,8 +646,8 @@ async function runDriverLoop(input: { }; } - if (next.kind === "clear" || next.kind === "compact") { - action = await runTurn({ kind: next.kind }); + if (next.kind === "clear" || next.kind === "compact" || next.kind === "restore-history") { + action = await runTurn(next); continue; } diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index 126dd40d4a..0968d3735e 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -91,6 +91,7 @@ import { resolveEffectiveAgentRuntime } from "#execution/effective-agent-config. import { reconcileSessionContinuationToken } from "#execution/reconcile-session-continuation-token.js"; import { hydrateDurableSession, refreshSessionFromTurnAgent } from "#execution/session.js"; import { createExecutionHistoryView } from "#execution/history-view.js"; +import { ResponseReleaseEventGate } from "#execution/response-release-event-gate.js"; import { resolveRuntimeCompiledArtifactsVersionedCacheKey } from "#runtime/cache-key.js"; import { createWorkflowRuntime } from "#execution/workflow-runtime.js"; import { bindDynamicConnections } from "#execution/dynamic-connections.js"; @@ -367,6 +368,12 @@ export async function turnStep(rawInput: TurnStepInput): Promise => { @@ -381,6 +388,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise => { + if (responseReleaseGate.intercept(event)) return; // A remote task's parent owns its HITL. Forward blocking events over // the task callback and keep them out of the child's local channel; // otherwise two TUIs can present and answer the same request. @@ -436,8 +444,6 @@ export async function turnStep(rawInput: TurnStepInput): Promise { + it("retains the exact prefix before the requested index", () => { + expect(restoreSessionHistory(session(), 1).history).toEqual([ + { content: "first", role: "user" }, + ]); + }); + + it.each([-1, 1.5, 3, Number.NaN])("rejects invalid index %s", (index) => { + expect(() => restoreSessionHistory(session(), index)).toThrow( + "History restoration index must be an integer from 0 through 2", + ); + }); +}); diff --git a/packages/eve/src/harness/history-restoration.ts b/packages/eve/src/harness/history-restoration.ts new file mode 100644 index 0000000000..02e0bdbefb --- /dev/null +++ b/packages/eve/src/harness/history-restoration.ts @@ -0,0 +1,16 @@ +import type { HarnessSession } from "#harness/types.js"; + +/** Restores the exact model-history prefix ending at `index`. */ +export function restoreSessionHistory(session: HarnessSession, index: number): HarnessSession { + validateHistoryRestoreIndex(session.history.length, index); + return { ...session, history: session.history.slice(0, index) }; +} + +/** Validates an index used as the exclusive end of a model-history prefix. */ +export function validateHistoryRestoreIndex(historyLength: number, index: number): void { + if (!Number.isInteger(index) || index < 0 || index > historyLength) { + throw new RangeError( + `History restoration index must be an integer from 0 through ${historyLength}; received ${String(index)}.`, + ); + } +} diff --git a/packages/eve/src/harness/tool-loop.test.ts b/packages/eve/src/harness/tool-loop.test.ts index 821ccff57c..e207a5fe81 100644 --- a/packages/eve/src/harness/tool-loop.test.ts +++ b/packages/eve/src/harness/tool-loop.test.ts @@ -915,6 +915,40 @@ describe("createToolLoopHarness", () => { ]); }); + it("restores the requested history prefix before a turn settles", async () => { + setupMockAgent({ + finishReason: "stop", + response: { messages: [{ content: "Sensitive draft", role: "assistant" }] }, + text: "Sensitive draft", + toolCalls: [], + toolResults: [], + }); + + const beforeResponseRelease = vi.fn().mockResolvedValue(1); + const runStep = createToolLoopHarness( + createTestConfig("conversation", undefined, { + beforeResponseRelease, + }), + ); + const previous = { content: "Earlier context", role: "user" as const }; + + const result = await runStep(createTestSession({ history: [previous] }), { + message: "Create a response", + }); + + expect(beforeResponseRelease).toHaveBeenCalledWith({ + history: [ + previous, + { content: "Create a response", role: "user" }, + { content: "Sensitive draft", role: "assistant" }, + ], + output: "Sensitive draft", + turnId: "", + }); + expect(result.session.history).toEqual([previous]); + expect(result.settledTurn).toEqual({ output: "" }); + }); + it("omits user messages with no model-visible content", async () => { setupMockAgent({ finishReason: "stop", @@ -2654,7 +2688,10 @@ describe("createToolLoopHarness", () => { }); const { emit, events } = createEventCollector(); - const runStep = createToolLoopHarness(createTestConfig("conversation", emit)); + const beforeResponseRelease = vi.fn().mockResolvedValue(undefined); + const runStep = createToolLoopHarness( + createTestConfig("conversation", emit, { beforeResponseRelease }), + ); const session = createTestSession({ outputSchema: { type: "object" } }); const result = await runStep(session, { message: "Hi" }); @@ -2682,6 +2719,7 @@ describe("createToolLoopHarness", () => { }), ); expect(result.session.outputSchema).toBeUndefined(); + expect(beforeResponseRelease).not.toHaveBeenCalled(); }); it("returns only the final assistant reply when a completed task step includes tool work", async () => { @@ -9564,6 +9602,26 @@ describe("createToolLoopHarness", () => { expect(stepViews).toEqual([compactedHistory]); }); + it("restores history without running a model turn", async () => { + const { emit, events } = createEventCollector(); + const runStep = createToolLoopHarness( + createTestConfig("conversation", emit, { restoreHistoryTo: 1 }), + ); + const session = createTestSession({ + history: [ + { content: "keep", role: "user" }, + { content: "remove", role: "assistant" }, + ], + }); + + const result = await runStep(session); + + expect(result.next).toBeNull(); + expect(result.session.history).toEqual([{ content: "keep", role: "user" }]); + expect(getCompatibilityEventTypes(events)).toEqual(["session.waiting"]); + expect(ToolLoopAgent).not.toHaveBeenCalled(); + }); + it("clears static and dynamic user instructions without rerunning lifecycle events", async () => { const { emit, events } = createEventCollector(); const resolveModel = vi.fn(); @@ -9592,7 +9650,6 @@ describe("createToolLoopHarness", () => { outputSchema, state, }); - const result = await runStep(session); expect(result.next).toBeNull(); diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index a731ead109..46b5cae344 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -169,6 +169,7 @@ import { resolveAssistantStepText, } from "#harness/messages.js"; import { normalizeProviderToolHistory } from "#harness/provider-tool-history.js"; +import { restoreSessionHistory } from "#harness/history-restoration.js"; import { getSupersededAuthorizationChallenges, setPendingAuthorization, @@ -548,6 +549,12 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { }); }; + if (config.restoreHistoryTo !== undefined) { + session = restoreSessionHistory(session, config.restoreHistoryTo); + await emit?.(createSessionWaitingEvent()); + return { next: null, session }; + } + if (config.clearOnly === true) { session = { ...session, @@ -2720,6 +2727,7 @@ async function handleStepResult(input: { } return finishConversationTurn({ + beforeResponseRelease: config.beforeResponseRelease, emissionState, emit, history: promptMessages, @@ -2843,6 +2851,7 @@ async function finishTaskTurn(input: { * ends the turn and the session waits for the next message. */ async function finishConversationTurn(input: { + readonly beforeResponseRelease?: ToolLoopHarnessConfig["beforeResponseRelease"]; readonly emissionState: ReturnType; readonly emit?: ToolLoopHarnessConfig["handleEvent"]; readonly history: readonly ModelMessage[]; @@ -2856,12 +2865,13 @@ async function finishConversationTurn(input: { session = clearTurnClientContextState(session); if (schema === undefined) { - if (emit) { - emissionState = await emitTurnEpilogue(emit, emissionState, "conversation"); - session = setHarnessEmissionState(session, emissionState); - } - const settledTurn = { output: stepOutput ?? "" } satisfies SettledTurn; - return { next: null, session, settledTurn }; + return settleConversationCandidate({ + beforeResponseRelease: input.beforeResponseRelease, + emissionState, + emit, + output: stepOutput ?? "", + session, + }); } const structured = extractFinalOutput(result); @@ -2884,12 +2894,48 @@ async function finishConversationTurn(input: { } session = persistStructuredAssistantTurn(session, history, structured); - if (emit) { - emissionState = await emitStructuredResult(emit, emissionState, structured, "conversation"); + return settleConversationCandidate({ + beforeResponseRelease: input.beforeResponseRelease, + emissionState, + emit, + emitAccepted: (emitFn, state) => + emitStructuredResult(emitFn, state, structured, "conversation"), + output: structured, + session, + }); +} + +async function settleConversationCandidate(input: { + readonly beforeResponseRelease?: ToolLoopHarnessConfig["beforeResponseRelease"]; + readonly emissionState: ReturnType; + readonly emit?: ToolLoopHarnessConfig["handleEvent"]; + readonly emitAccepted?: ( + emit: NonNullable, + state: ReturnType, + ) => Promise>; + readonly output: unknown; + readonly session: HarnessSession; +}): Promise { + let { emissionState, session } = input; + const restoreHistoryTo = await input.beforeResponseRelease?.({ + history: session.history, + output: input.output, + turnId: emissionState.turnId, + }); + const restored = restoreHistoryTo !== undefined; + if (restored) session = restoreSessionHistory(session, restoreHistoryTo); + if (input.emit) { + emissionState = restored + ? await emitTurnEpilogue(input.emit, emissionState, "conversation") + : await (input.emitAccepted?.(input.emit, emissionState) ?? + emitTurnEpilogue(input.emit, emissionState, "conversation")); session = setHarnessEmissionState(session, emissionState); } - const settledTurn = { output: structured } satisfies SettledTurn; - return { next: null, session, settledTurn }; + return { + next: null, + session, + settledTurn: { output: restored ? "" : input.output }, + }; } /** Replays a parked dynamic workflow with completed child-agent results. */ diff --git a/packages/eve/src/harness/types.ts b/packages/eve/src/harness/types.ts index e4fde11576..1848c3f368 100644 --- a/packages/eve/src/harness/types.ts +++ b/packages/eve/src/harness/types.ts @@ -289,6 +289,8 @@ export interface ToolLoopHarnessConfig { readonly clearOnly?: boolean; /** Forces one context-compaction pass without running a model turn. */ readonly compactOnly?: boolean; + /** Restores model history to an earlier index without running a model turn. */ + readonly restoreHistoryTo?: number; /** * Exposes the `Workflow` orchestration tool — an isolated JavaScript sandbox * whose only callable operations are this agent's subagents and remote @@ -305,6 +307,12 @@ export interface ToolLoopHarnessConfig { */ readonly workflowMaxSubagents?: number; readonly handleEvent?: HandleEventFn; + /** Optional history restoration requested before terminal turn release. */ + readonly beforeResponseRelease?: (input: { + readonly history: readonly ModelMessage[]; + readonly output: unknown; + readonly turnId: string; + }) => Promise; /** Projects raw durable history before it crosses a message-bearing boundary. */ readonly historyProjector?: HistoryViewProjector; /** Execution-prepared view of the history supplied to the first harness step. */ diff --git a/packages/eve/src/internal/testing/mocks/mock-channel-operations.ts b/packages/eve/src/internal/testing/mocks/mock-channel-operations.ts index ab821c5a32..327bafea95 100644 --- a/packages/eve/src/internal/testing/mocks/mock-channel-operations.ts +++ b/packages/eve/src/internal/testing/mocks/mock-channel-operations.ts @@ -43,6 +43,9 @@ export function mockChannelContext( async clear() { return { status: "no_active_session" } as never; }, + async restoreHistory() { + return { status: "no_active_session" } as never; + }, async reset() { return { status: "no_active_session" } as never; }, diff --git a/packages/eve/src/protocol/restore-history.test.ts b/packages/eve/src/protocol/restore-history.test.ts new file mode 100644 index 0000000000..8662784dd8 --- /dev/null +++ b/packages/eve/src/protocol/restore-history.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "vitest"; + +import { + RestoreHistoryRequestSchema, + RestoreHistoryResponseSchema, +} from "#protocol/restore-history.js"; + +describe("history restoration protocol", () => { + it("accepts a nonnegative history index", () => { + expect(RestoreHistoryRequestSchema.parse({ to: 0 })).toEqual({ to: 0 }); + }); + + it.each([-1, 1.5, "1"])("rejects invalid history index %s", (to) => { + expect(RestoreHistoryRequestSchema.safeParse({ to }).success).toBe(false); + }); + + it("accepts both successful outcomes", () => { + expect( + RestoreHistoryResponseSchema.parse({ + ok: true, + sessionId: "sess_1", + status: "accepted", + }), + ).toEqual({ ok: true, sessionId: "sess_1", status: "accepted" }); + expect(RestoreHistoryResponseSchema.parse({ ok: true, status: "no_active_session" })).toEqual({ + ok: true, + status: "no_active_session", + }); + }); +}); diff --git a/packages/eve/src/protocol/restore-history.ts b/packages/eve/src/protocol/restore-history.ts new file mode 100644 index 0000000000..28ed8bb243 --- /dev/null +++ b/packages/eve/src/protocol/restore-history.ts @@ -0,0 +1,29 @@ +import { z } from "#compiled/zod/index.js"; + +/** Request body for restoring an exact model-history snapshot prefix. */ +export const RestoreHistoryRequestSchema = z.object({ + to: z.number().int().nonnegative(), +}); + +export type RestoreHistoryRequest = z.infer; + +/** Outcome of queueing model-history restoration. */ +export type RestoreHistoryStatus = "accepted" | "no_active_session"; + +/** Successful response returned by the session history-restoration route. */ +export type RestoreHistoryResponse = + | { readonly ok: true; readonly sessionId: string; readonly status: "accepted" } + | { readonly ok: true; readonly status: "no_active_session" }; + +/** Validates successful session history-restoration responses. */ +export const RestoreHistoryResponseSchema: z.ZodType = z.discriminatedUnion( + "status", + [ + z.object({ + ok: z.literal(true), + sessionId: z.string().min(1), + status: z.literal("accepted"), + }), + z.object({ ok: z.literal(true), status: z.literal("no_active_session") }), + ], +); diff --git a/packages/eve/src/protocol/routes.ts b/packages/eve/src/protocol/routes.ts index d85aa81387..1b4a900cee 100644 --- a/packages/eve/src/protocol/routes.ts +++ b/packages/eve/src/protocol/routes.ts @@ -34,6 +34,9 @@ export const EVE_SESSION_COMPACT_ROUTE_PATTERN = `${EVE_SESSION_ROUTE_PATH}/:ses /** Stable route pattern for clearing one exact session ID. */ export const EVE_SESSION_CLEAR_ROUTE_PATTERN = `${EVE_SESSION_ROUTE_PATH}/:sessionId/clear`; +/** Stable route pattern for restoring one exact session's model history. */ +export const EVE_SESSION_RESTORE_HISTORY_ROUTE_PATTERN = `${EVE_SESSION_ROUTE_PATH}/:sessionId/restore-history`; + /** Stable route pattern for resetting one exact session ID. */ export const EVE_SESSION_RESET_ROUTE_PATTERN = `${EVE_SESSION_ROUTE_PATH}/:sessionId/reset`; @@ -153,6 +156,11 @@ export function createEveSessionClearRoutePath(sessionId: string): string { return `${EVE_SESSION_ROUTE_PATH}/${encodeURIComponent(sessionId)}/clear`; } +/** Builds the ID-addressed history-restoration route for one session. */ +export function createEveSessionRestoreHistoryRoutePath(sessionId: string): string { + return `${EVE_SESSION_ROUTE_PATH}/${encodeURIComponent(sessionId)}/restore-history`; +} + /** Builds the ID-addressed reset route for one session. */ export function createEveSessionResetRoutePath(sessionId: string): string { return `${EVE_SESSION_ROUTE_PATH}/${encodeURIComponent(sessionId)}/reset`; diff --git a/packages/eve/src/public/channels/eve-session-routes.test.ts b/packages/eve/src/public/channels/eve-session-routes.test.ts index f9b07da5ce..399a5ae8e5 100644 --- a/packages/eve/src/public/channels/eve-session-routes.test.ts +++ b/packages/eve/src/public/channels/eve-session-routes.test.ts @@ -23,6 +23,7 @@ function createFixedSession(overrides: Partial = {}): Session { cancel: vi.fn().mockResolvedValue({ sessionId: "wrun_A", status: "accepted" }), compact: vi.fn().mockResolvedValue({ sessionId: "wrun_A", status: "accepted" }), clear: vi.fn().mockResolvedValue({ sessionId: "wrun_A", status: "accepted" }), + restoreHistory: vi.fn().mockResolvedValue({ sessionId: "wrun_A", status: "accepted" }), reset: vi.fn().mockResolvedValue({ previousSessionId: "wrun_A", status: "reset" }), getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), getStreamTailIndex: vi.fn().mockResolvedValue(-1), @@ -220,6 +221,31 @@ describe("eve ID-addressed session routes", () => { expect(session[operation]).toHaveBeenCalledTimes(1); }); + it("validates and dispatches history restoration", async () => { + const session = createFixedSession(); + const handler = route("POST", "/eve/v1/session/:sessionId/restore-history"); + const response = await handler( + new Request("https://eve.test/eve/v1/session/wrun_A/restore-history", { + body: JSON.stringify({ to: 2 }), + method: "POST", + }), + createArgs(session), + ); + + expect(response.status).toBe(202); + expect(session.restoreHistory).toHaveBeenCalledWith({ to: 2 }); + + const invalid = await handler( + new Request("https://eve.test/eve/v1/session/wrun_A/restore-history", { + body: JSON.stringify({ to: -1 }), + method: "POST", + }), + createArgs(session), + ); + expect(invalid.status).toBe(400); + expect(session.restoreHistory).toHaveBeenCalledTimes(1); + }); + it("forwards owned-task cancellation without changing the response", async () => { const session = createFixedSession(); const response = await route("POST", "/eve/v1/session/:sessionId/cancel")( diff --git a/packages/eve/src/public/channels/eve-subagent-stream.test.ts b/packages/eve/src/public/channels/eve-subagent-stream.test.ts index 5d93a2a91a..2f9151e420 100644 --- a/packages/eve/src/public/channels/eve-subagent-stream.test.ts +++ b/packages/eve/src/public/channels/eve-subagent-stream.test.ts @@ -221,6 +221,9 @@ function createHarness(input: { async clear() { return { sessionId: coordinates.parentSessionId, status: "accepted" }; }, + async restoreHistory() { + return { sessionId: coordinates.parentSessionId, status: "accepted" }; + }, async reset() { return { previousSessionId: coordinates.parentSessionId, status: "reset" }; }, diff --git a/packages/eve/src/public/channels/eve.test.ts b/packages/eve/src/public/channels/eve.test.ts index 35cfa0f362..75d1fb67c6 100644 --- a/packages/eve/src/public/channels/eve.test.ts +++ b/packages/eve/src/public/channels/eve.test.ts @@ -75,6 +75,10 @@ function createMockSession(overrides: Partial = {}): Session { cancel: vi.fn().mockResolvedValue({ status: "no_active_turn" }), compact: vi.fn().mockResolvedValue({ sessionId: "test-session-id", status: "accepted" }), clear: vi.fn().mockResolvedValue({ sessionId: "test-session-id", status: "accepted" }), + restoreHistory: vi.fn().mockResolvedValue({ + sessionId: "test-session-id", + status: "accepted", + }), reset: vi.fn().mockResolvedValue({ previousSessionId: "test-session-id", status: "reset", diff --git a/packages/eve/src/public/channels/slack/session-operations.ts b/packages/eve/src/public/channels/slack/session-operations.ts index fc7d0e6022..fa849b2a7f 100644 --- a/packages/eve/src/public/channels/slack/session-operations.ts +++ b/packages/eve/src/public/channels/slack/session-operations.ts @@ -32,6 +32,7 @@ export interface SlackSessionOperations { cancel(options?: { readonly turnId?: string }): ReturnType; compact(): ReturnType; clear(): ReturnType; + restoreHistory(input: { readonly to: number }): ReturnType; reset(options?: { readonly reason?: string }): ReturnType; resolveSession(): ReturnType; } @@ -71,6 +72,9 @@ export function bindSlackSessionOperations(input: { async clear() { return await source.clear(); }, + async restoreHistory(restoreInput) { + return await source.restoreHistory(restoreInput); + }, async reset(options) { return await source.reset(options); }, diff --git a/packages/eve/src/public/definitions/channel.test.ts b/packages/eve/src/public/definitions/channel.test.ts index 3e18f29ce8..af161680f7 100644 --- a/packages/eve/src/public/definitions/channel.test.ts +++ b/packages/eve/src/public/definitions/channel.test.ts @@ -588,6 +588,9 @@ describe("defineChannel", () => { async clear() { return { sessionId: channelId, status: "accepted" as const }; }, + async restoreHistory() { + return { sessionId: channelId, status: "accepted" as const }; + }, async reset() { return { previousSessionId: channelId, status: "reset" as const }; }, diff --git a/packages/eve/src/public/definitions/hook.ts b/packages/eve/src/public/definitions/hook.ts index 48e2bc3b5a..4ef639f15a 100644 --- a/packages/eve/src/public/definitions/hook.ts +++ b/packages/eve/src/public/definitions/hook.ts @@ -1,3 +1,4 @@ +import type { ModelMessage } from "ai"; import type { HandleMessageStreamEvent } from "../../protocol/message.js"; import type { SessionContext } from "./callback-context.js"; import type { ExactDefinition } from "./exact.js"; @@ -98,6 +99,30 @@ export type StreamEventHooks = { readonly [TKey_ in TKey]?: StreamEventHook>; }; +/** Model history available to a response-release hook. */ +export interface ResponseReleaseHistory { + readonly messages: readonly ModelMessage[]; + /** Requests restoration to the exact prefix ending before `index`. */ + restoreTo(index: number): void; +} + +/** Candidate conversation turn presented immediately before settlement. */ +export interface ResponseReleaseCandidate { + readonly history: ResponseReleaseHistory; + readonly output: unknown; + readonly turnId: string; +} + +/** + * Runs after final synthesis but before terminal content reaches the channel. + * Requesting history restoration suppresses the pending terminal completion. + * Earlier stream events and side effects are not retracted. + */ +export type BeforeResponseReleaseHook = ( + candidate: ResponseReleaseCandidate, + ctx: HookContext, +) => void | Promise; + /** * Public hook definition authored in `agent/hooks/*.ts`. * @@ -108,6 +133,7 @@ export type StreamEventHooks = { * `defineInstructions` in `agent/instructions/`. */ export interface HookDefinition { + readonly beforeResponseRelease?: BeforeResponseReleaseHook; readonly events?: StreamEventHooks; } diff --git a/packages/eve/src/public/hooks/index.ts b/packages/eve/src/public/hooks/index.ts index aeeb55dce2..757d50ee8b 100644 --- a/packages/eve/src/public/hooks/index.ts +++ b/packages/eve/src/public/hooks/index.ts @@ -7,6 +7,7 @@ */ export { + type BeforeResponseReleaseHook, type HookContext, type HookDefinition, type HookEvent, @@ -15,5 +16,7 @@ export { type HookEventType, type StreamEventHook, type StreamEventHooks, + type ResponseReleaseCandidate, + type ResponseReleaseHistory, defineHook, } from "#public/definitions/hook.js"; diff --git a/packages/eve/src/runtime/hooks/registry.test.ts b/packages/eve/src/runtime/hooks/registry.test.ts index a81e2fe25d..c6f72a6d48 100644 --- a/packages/eve/src/runtime/hooks/registry.test.ts +++ b/packages/eve/src/runtime/hooks/registry.test.ts @@ -4,17 +4,22 @@ import type { ResolvedHookDefinition } from "../types.js"; import { createEmptyHookRegistry, createRuntimeHookRegistry } from "./registry.js"; describe("createRuntimeHookRegistry", () => { - it("splits typed and wildcard stream-event subscribers", () => { + it("splits settlement and stream-event hooks", () => { + const beforeResponseRelease = async () => undefined; const typed = async () => {}; const wildcard = async () => {}; const registry = createRuntimeHookRegistry([ makeHook({ + beforeResponseRelease, slug: "audit", events: { "message.completed": typed, "*": wildcard }, }), ]); + expect(registry.beforeResponseRelease).toEqual([ + { handler: beforeResponseRelease, slug: "audit" }, + ]); expect( (registry.streamEventsByType.get("message.completed") ?? []).map((e) => e.eventType), ).toEqual(["message.completed"]); @@ -25,16 +30,19 @@ describe("createRuntimeHookRegistry", () => { describe("createEmptyHookRegistry", () => { it("returns flat empty buckets", () => { const registry = createEmptyHookRegistry(); + expect(registry.beforeResponseRelease).toEqual([]); expect(registry.streamEventsByType.size).toBe(0); expect(registry.streamEventsWildcard).toEqual([]); }); }); function makeHook(partial: { + readonly beforeResponseRelease?: ResolvedHookDefinition["beforeResponseRelease"]; readonly slug: string; readonly events?: ResolvedHookDefinition["events"]; }): ResolvedHookDefinition { return { + beforeResponseRelease: partial.beforeResponseRelease, events: partial.events ?? {}, exportName: undefined, logicalPath: `hooks/${partial.slug}.ts`, diff --git a/packages/eve/src/runtime/hooks/registry.ts b/packages/eve/src/runtime/hooks/registry.ts index 99db43c211..fa60290bba 100644 --- a/packages/eve/src/runtime/hooks/registry.ts +++ b/packages/eve/src/runtime/hooks/registry.ts @@ -1,5 +1,5 @@ import type { MessageStreamEvent } from "#protocol/message.js"; -import type { StreamEventHook } from "../../public/definitions/hook.js"; +import type { BeforeResponseReleaseHook, StreamEventHook } from "../../public/definitions/hook.js"; import type { ResolvedHookDefinition } from "../types.js"; /** @@ -8,6 +8,11 @@ import type { ResolvedHookDefinition } from "../types.js"; * `eventType` is `"*"` for wildcard subscribers, otherwise the typed * event name. */ +interface RuntimeBeforeResponseReleaseHookEntry { + readonly handler: BeforeResponseReleaseHook; + readonly slug: string; +} + interface RuntimeStreamEventHookEntry { readonly slug: string; readonly handler: StreamEventHook; @@ -21,6 +26,7 @@ interface RuntimeStreamEventHookEntry { * without scanning every entry. */ export interface RuntimeHookRegistry { + readonly beforeResponseRelease: readonly RuntimeBeforeResponseReleaseHookEntry[]; readonly streamEventsByType: ReadonlyMap; readonly streamEventsWildcard: readonly RuntimeStreamEventHookEntry[]; } @@ -33,6 +39,7 @@ export interface RuntimeHookRegistry { */ export function createEmptyHookRegistry(): RuntimeHookRegistry { return { + beforeResponseRelease: [], streamEventsByType: new Map(), streamEventsWildcard: [], }; @@ -49,10 +56,14 @@ export function createEmptyHookRegistry(): RuntimeHookRegistry { export function createRuntimeHookRegistry( resolvedHooks: readonly ResolvedHookDefinition[], ): RuntimeHookRegistry { + const beforeResponseRelease: RuntimeBeforeResponseReleaseHookEntry[] = []; const streamEventsByType = new Map(); const streamEventsWildcard: RuntimeStreamEventHookEntry[] = []; for (const hook of resolvedHooks) { + if (hook.beforeResponseRelease !== undefined) { + beforeResponseRelease.push({ handler: hook.beforeResponseRelease, slug: hook.slug }); + } for (const [eventType, handler] of Object.entries(hook.events)) { const entry: RuntimeStreamEventHookEntry = { slug: hook.slug, handler, eventType }; if (eventType === "*") { @@ -66,6 +77,7 @@ export function createRuntimeHookRegistry( } return { + beforeResponseRelease, streamEventsByType, streamEventsWildcard, }; diff --git a/packages/eve/src/runtime/resolve-hook.test.ts b/packages/eve/src/runtime/resolve-hook.test.ts index 8abf050e58..b860e92d71 100644 --- a/packages/eve/src/runtime/resolve-hook.test.ts +++ b/packages/eve/src/runtime/resolve-hook.test.ts @@ -48,6 +48,17 @@ describe("resolveHookDefinition", () => { expect(Object.keys(resolved.events).sort()).toEqual(["*", "message.completed"]); }); + it("resolves a pre-settlement turn hook", async () => { + const definition = buildDefinition({ slug: "review" }); + const beforeResponseRelease = () => undefined; + const moduleMap = buildModuleMap(definition.sourceId, { + default: { beforeResponseRelease }, + }); + + const resolved = await resolveHookDefinition(definition, moduleMap, undefined); + expect(resolved.beforeResponseRelease).toBe(beforeResponseRelease); + }); + it("accepts a hook with only `events` declared", async () => { const definition = buildDefinition({ slug: "audit" }); const moduleMap = buildModuleMap(definition.sourceId, { diff --git a/packages/eve/src/runtime/resolve-hook.ts b/packages/eve/src/runtime/resolve-hook.ts index 2fffa508fc..8afc098db7 100644 --- a/packages/eve/src/runtime/resolve-hook.ts +++ b/packages/eve/src/runtime/resolve-hook.ts @@ -2,7 +2,7 @@ import type { CompiledHookDefinition } from "../compiler/manifest.js"; import type { CompiledModuleMap } from "../compiler/module-map.js"; import { expectFunction, expectObjectRecord } from "../internal/authored-module.js"; import type { MessageStreamEvent } from "../protocol/message.js"; -import type { StreamEventHook } from "../public/definitions/hook.js"; +import type { BeforeResponseReleaseHook, StreamEventHook } from "../public/definitions/hook.js"; import { toErrorMessage } from "../shared/errors.js"; import { loadResolvedModuleExport, ResolveAgentError } from "./resolve-helpers.js"; import type { ResolvedHookDefinition } from "./types.js"; @@ -33,6 +33,14 @@ export async function resolveHookDefinition( describe(definition, "to return an object"), ); + const beforeResponseReleaseRaw = resolvedRecord.beforeResponseRelease; + const beforeResponseRelease = + beforeResponseReleaseRaw === undefined + ? undefined + : (expectFunction( + beforeResponseReleaseRaw, + describe(definition, "to provide a function for beforeResponseRelease"), + ) as BeforeResponseReleaseHook); const events: Record> = {}; const eventsRaw = resolvedRecord.events; @@ -52,6 +60,7 @@ export async function resolveHookDefinition( } return { + beforeResponseRelease, events, exportName: definition.exportName, logicalPath: definition.logicalPath, diff --git a/packages/eve/src/runtime/types.ts b/packages/eve/src/runtime/types.ts index f83c67f48a..ae9c69d38a 100644 --- a/packages/eve/src/runtime/types.ts +++ b/packages/eve/src/runtime/types.ts @@ -211,6 +211,7 @@ export type ResolvedToolDefinition = Readonly< * the resolved maps. */ export interface ResolvedHookDefinition extends ResolvedModuleSourceRef { + readonly beforeResponseRelease?: import("#public/definitions/hook.js").BeforeResponseReleaseHook; /** * Path-relative slug used for diagnostics and ordering. */ diff --git a/packages/eve/test/eve-run-stream-channel.test.ts b/packages/eve/test/eve-run-stream-channel.test.ts index 69178219e7..6beb188e9c 100644 --- a/packages/eve/test/eve-run-stream-channel.test.ts +++ b/packages/eve/test/eve-run-stream-channel.test.ts @@ -186,6 +186,9 @@ function createMockAttachSession(events: ReadableStream) { async clear() { return { sessionId: "session_xyz", status: "accepted" }; }, + async restoreHistory() { + return { sessionId: "session_xyz", status: "accepted" }; + }, async reset() { return { previousSessionId: "session_xyz", status: "reset" }; }, diff --git a/research/sensitive-response-review-gates.md b/research/sensitive-response-review-gates.md new file mode 100644 index 0000000000..780a6e71ad --- /dev/null +++ b/research/sensitive-response-review-gates.md @@ -0,0 +1,588 @@ +--- +issue: TBD +status: draft +last_updated: "2026-09-04" +--- + +# Sensitive response review gates + +> **AI status:** Written entirely by AI; human review pending. + +## Summary + +Current eve can securely approve each sensitive tool call before it executes, or isolate sensitive +work in a subagent or workflow tool, but it cannot pause a root turn after synthesis and before both +canonical history commit and channel delivery. Authored hooks cannot provide that boundary: channel +handlers run first, the event is durably recorded next, and hooks are observe-only subscribers after +both. Tool results also enter the turn's durable working snapshot after each model step, before a +later model call synthesizes the final answer. + +The immediate framework direction is a general serialized `restoreHistory` session control plus a +minimal `beforeResponseRelease` hook. The control restores an exact model-history prefix. The hook lets authored policy inspect terminal output while eve holds its +`message.completed` event and request the same restoration before release. This supports the narrow +use case without adding a framework notion of sensitivity or a lifecycle unit spanning several +turns. + +This thin feature is logical restoration, not confidential execution. A stronger framework-level +**quarantined turn** remains the direction when unapproved content must also stay outside ordinary +events, hooks, memory, telemetry, and execution storage. + +Until that framework feature exists, the best approximation is to put sensitive retrieval and +synthesis behind one declared subagent invoked by a waiting workflow tool, keep sensitive tools off +the root, and have the workflow obtain a private approval before directly delivering the exact +approved draft. Return only a non-sensitive receipt to the root. This gives one review prompt and +keeps sensitive content out of root history, but it needs a bespoke private review surface and still +persists sensitive data in the child/workflow or an external protected store. Existing Slack HITL is +posted to the session thread and is not itself a private review enclave. + +A durable review cannot guarantee that sensitive content is never persisted anywhere: the model +needs tool results across steps, and the draft must survive a process restart while awaiting a +human. The achievable contract is that unapproved content is persisted only in an access-controlled, +retention-bounded quarantine, never in canonical session history, ordinary event streams, channel +messages, unprivileged hooks, or content-bearing telemetry. + +## Implementation direction + +Implementation exploration tested a per-turn checkpoint and a pre-commit `"commit" | "rollback"` +hook. That design is too tightly coupled to the current turn. Existing HITL parks and resumes across +turn boundaries, while the application may need to remove all model history from the original user +question through final synthesis. + +The narrower plan separates history restoration from terminal release gating. It does not add a new +first-class lifecycle unit or make sensitivity a framework concept. + +### Serialized session history restoration + +Add a general session control alongside `clear` and `compact` that restores model-visible history to +an earlier boundary in the current serialized session history. + +```ts +await session.restoreHistory({ to: questionIndex }); +``` + +The control is serialized through the existing session inbox. It may retain only an exact prefix of +the observed history; authored code cannot replace or inject arbitrary messages. The runtime rejects an out-of-range index. +Like `clear` and `compact`, restoration affects future model-visible history. It does not retract +stream events, provider calls, tool side effects, memory writes, traces, sandbox writes, external +records, notifications, or channel messages already observed. + +The initial implementation should keep this mechanism independent of task cancellation. An index +into model history does not identify every task or side effect created while the removed suffix was +produced. Adding that ownership requires a separate, explicit contract rather than inferring it from +a per-turn checkpoint. + +### Minimal pre-release boundary + +History restoration alone cannot prevent an already-completed answer from reaching Slack. Keep a +minimal `beforeResponseRelease` hook for policies that must inspect the terminal candidate while eve holds +its terminal `message.completed` event. + +The hook does not return a framework-prescribed commit or rollback decision. It may request history +restoration through a settlement-scoped capability: + +```ts +export default defineHook({ + beforeResponseRelease(candidate) { + const questionIndex = findRelevantUserMessage(candidate.history.messages); + const attempt = candidate.history.messages.slice(questionIndex); + + if (containsSensitiveData(attempt) && !containsRequiredApproval(attempt)) { + candidate.history.restoreTo(questionIndex); + } + }, +}); +``` + +If no hook requests restoration, eve keeps the candidate history and releases the terminal event. +If a hook requests restoration, eve applies the validated prefix restoration and suppresses the +pending terminal event. Multiple hooks may request restoration; the earliest valid boundary wins. +Throwing fails settlement rather than releasing content. + +The settlement capability and serialized session control share one internal history-restoration +operation, but their public timing remains explicit: + +- `session.restoreHistory(...)` is a serialized control for an idle or concurrently addressed + session. It changes future model context but cannot retract prior delivery. +- `candidate.history.restoreTo(...)` is available only while the terminal event is held. Restoration + at this boundary also suppresses that pending event. + +### Policy inputs and limits + +The authored policy still needs trustworthy evidence. Raw model history may contain only a tool's +`toModelOutput` projection, so applications must preserve any sensitivity label needed by the hook. +Approval must be correlated with the candidate output or its digest; the mere presence of an earlier +approval is insufficient. + +This design provides logical history restoration and terminal response suppression, not confidential +execution. Earlier events, hooks, memory, telemetry, execution storage, and external systems may +already have observed candidate content. The quarantined-turn design below remains necessary when +unapproved content must stay outside those internal consumers. Private Slack preview is orthogonal: +it controls who can see approval UI, while restoration controls future model context and the held +terminal response. + +## Required security contract + +The request combines four distinct guarantees that should not be conflated: + +1. **Retrieval authorization:** the caller may access the resource returned by each tool. This stays + in the tool and is keyed to the requested resource; HITL is not authorization. +2. **Information-flow tracking:** a trusted runtime decision marks the turn sensitive when any tool + result is sensitive. A model statement or prompt instruction is not a security boundary. +3. **Private review:** only an authenticated, policy-authorized reviewer can see and decide on the + candidate answer. +4. **Atomic release:** no unreviewed content reaches canonical history or an outward channel, and + the bytes posted after approval are the bytes reviewed. A second model synthesis after approval + breaks this guarantee. + +“Private” must name an audience. For Slack, a thread button is not private from thread participants. +The likely reviewer is the authenticated triggering user, with an optional response policy for a +separate approver. The preview should be an ephemeral message, DM, or authenticated review page, +not the public thread. + +## Current eve execution boundaries + +The relevant current behavior is: + +- Tool `approval` runs before `execute`. It can inspect input and session auth, but not the eventual + output or its sensitivity. `always()` therefore implements call-level authorization, not answer + review. +- A workflow tool can fetch data, call `ask`, suspend durably, and return only after approval. + Several fetches can be grouped into one workflow and one `ask`. +- `ask` emits `input.requested` on the owning session. Slack's default handler posts the prompt and + controls to the thread. The prompt and event are not an out-of-band private secret channel. +- Declared subagents have separate history, tools, connections, hooks, state, and sandbox. Parent + hooks do not observe child events. Child HITL and authorization events are nevertheless proxied to + the parent session so its channel can render them. +- Declared subagents are background tasks. The parent first receives a working receipt; completion + later wakes it with the result. The parent may continue and emit messages while the child works. +- A child parks after answering and remains resumable. There is no authored “discard this child and + erase its durable history” primitive. Task cancellation stops live work; it is not secure erase. +- A tool's `toModelOutput` can project a full result to a reduced model-visible value, but + `action.result` still carries the full output. It is useful for handle-based designs, not a + comprehensive confidentiality boundary. +- The harness appends tool-call response messages after every completed model step and uses that + snapshot on the next model call. Tool gathering and answer synthesis are therefore not one + uncommitted callback scope. +- The harness emits text deltas and `message.completed` while consuming a model response. Slack + normally posts terminal text from `message.completed`, but the durable stream, instrumentation, + and hooks can observe content earlier. Slack reasoning and action status handlers can also expose + derived progress unless disabled or redacted. +- Channel event handlers run before the event is stamped and written. Their errors are swallowed. + Authored hooks run only after the channel handler and durable stream write; return values are + ignored, and throwing turns the run into a failure rather than rolling it back. +- Session history is append-only within a turn. Compaction is the intentional rewrite mechanism; + public clear removes the whole model history. Selectively deleting sensitive tool results after + they have been committed would also risk invalid provider history by separating tool calls from + their required results. + +## Information-flow timelines + +Legend: + +```text +[M] main-agent model context/history [S] subagent model context/history +[Q] private quarantined durable state [C] canonical eve durable history/stream +[R] private reviewer surface [P] public/shared channel +! potentially sensitive content crosses this boundary +--- no sensitive content crosses this boundary +``` + +These timelines distinguish **model exposure** from **user/channel exposure**. Any model that +synthesizes from sensitive values necessarily receives them in its inference context. The review +gate controls later persistence and release; it cannot make that inference disappear. Provider-side +logging and retention must therefore be configured independently. + +### Current root tool loop without a release gate + +Multiple tool calls may span several model steps. Each completed step becomes the durable prompt for +the next one, so sensitive results enter canonical history before final synthesis. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +user request public tool sensitive tool final synthesis Slack + | | | | | + v v v v v +[M] request ───── [M] safe result ──! [M] sensitive result ──! [M] draft ───────! [P] draft + | | | | + v v v v +[C] request ───── [C] safe step ────! [C] sensitive step ───! [C] final answer + + No review or rollback boundary +``` + +Content can reach the channel before an authored hook sees it: + +```text +model text -> message.completed -> Slack channel handler posts ! -> durable event write -> hook +``` + +Text deltas, reasoning, tool events, instrumentation, and hooks may observe content even before the +terminal `message.completed` event, depending on their configuration. + +### Option A1: approval before every sensitive retrieval + +The sensitive result does not exist until approval, but several calls can produce several prompts. +Once approved, the result follows the ordinary root path and there is no answer-level review. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +sensitive call approval fetch result second call approval fetch result synthesize Slack + | | | | | | | | + v v v v v v v v +[M] call ─────── [R] gate ──! [M]/[C] result ─ [M] call ─ [R] gate ─! [M]/[C] result ─! draft ─! [P] + one prompt another prompt +``` + +### Option A2: retrieve, classify, then approve inside workflow tools + +Each workflow can withhold its result from the main model until approval. The value must still +survive the durable wait inside workflow state, and a normal `ask.prompt` that includes it is sent +through ordinary HITL events and the Slack thread. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +main calls tool workflow fetches durable wait/review return synthesis Slack + | | | | | | + v v v v v v +[M] call -----------! workflow private state ----! [R]* ------------! [M]/[C] result ─! draft ─! [P] + +* With current standard Slack HITL, this is normally a thread post, not a private reviewer surface. +``` + +Wrapping all retrieval and synthesis in one workflow reduces the interaction to one review, but +then the workflow—not the open-ended root loop—must own the complete answer. + +### Option B: confidential declared subagent + +The sensitive values and draft remain outside main-agent history while the child works. Returning +the candidate exposes it to the parent and commits it as a task/tool result. If the parent rewrites +it, the delivered answer is no longer the reviewed answer. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +parent delegates child fetches child synthesizes review child returns + | | | | | + v v v v v +[M] task receipt --- --- [S] request ─────! [S] results/draft ──! [R]* ───────────! [M]/[C] result + | + parent model rewrites ! ----------+ + | + v + ! [P] Slack + +* Child HITL is proxied to the parent channel. Putting the draft in the request leaks it to that + channel unless a new private renderer or external review surface resolves an opaque candidate id. +``` + +A safer variant never returns the candidate to the parent: + +```text +[S] sensitive work -> [R] approve exact draft -> direct idempotent Slack post ! [P] + | + +-> main receives only { delivered: true } --- [M]/[C] +``` + +That variant works as an interim application architecture but bypasses normal parent synthesis and +requires custom secure review and delivery code. + +### Option C implemented as a current authored hook + +This gate is too late. Earlier tool steps and the draft are already canonical, and the Slack handler +runs before the hook on `message.completed`. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +tool result committed draft completed Slack handler authored hook delete? + | | | | | + v v v v v +! [M]/[C] sensitive result ──! [M]/[C] draft ─────! [P] posted ──────── [R] too late ── X unsafe + +X Selective deletion can orphan tool calls, cannot retract Slack or event consumers, and does not + erase traces, hooks, provider logs, sandboxes, or external systems. +``` + +### Recommended framework lifecycle: quarantined turn + +The main agent may still perform the work, but the active branch is private and non-canonical until +review. Sensitivity is accumulated by trusted runtime metadata. Ordinary hooks, memory, streams, +and channel handlers receive only redacted lifecycle metadata before release. + +```text +time ───────────────────────────────────────────────────────────────────────────────▶ + +canonical checkpoint mixed tool loop in candidate terminal draft private review + | | | | | + v v v v v +[C] request ────────> [Q]/[M] safe ───! [Q]/[M] sensitive ─! [Q]/[M] draft ────! [R] candidate + | | | | | + | +----------+--------------------+ | + | [C]/ordinary stream: redacted metadata only | + | | + | approve exact bytes / approved edit | reject + | | | + | v v + +──────────────────────! [C] minimized approved answer destroy [Q] + | [C] safe status only + v + idempotent release outbox + | + v + ! [P] Slack +``` + +Crash-safe approval requires `[Q]` to be persisted somewhere protected. The guarantee is therefore +not “sensitive bytes are never persisted”; it is “unapproved bytes exist only in the quarantined +security domain and never enter canonical history or outward delivery.” Approval should normally +promote only the exact reviewed answer and provenance/audit metadata, not the raw tool branch. + +## Option analysis + +### Option A: approve sensitive tool results individually + +There are two variants: + +- `approval: always()` asks before retrieval. This is the strongest existing boundary when data + must not even be fetched without consent, but it cannot classify the returned value and may + produce many prompts. +- A waiting workflow tool can retrieve first, inspect the result, then call `ask` before returning + it to the model. This can make one tool call self-contained and durable, but each independently + called wrapper still asks separately. The retrieved value is already in the workflow's durable + execution record, and showing it in `ask.prompt` places it in normal HITL delivery. + +A handle-based variation returns opaque references from retrieval tools and exposes one final +“release these references” workflow tool. It batches consent and keeps raw values out of root model +history until release. It still cannot let the root model synthesize a draft before release; if a +human must review the answer rather than the source data, synthesis has to move behind the handle. + +**Use when:** approval is about access to each source or operation, or the sensitive data can be +summarized deterministically inside one workflow. + +**Limitation:** it does not naturally review one answer assembled by an open-ended root tool loop. + +### Option B: isolate sensitive work in a subagent + +This is feasible if the boundary is structural: + +- Remove sensitive tools from the root and expose them only to a declared subagent. +- Have the child perform every operation that needs the sensitive values and produce a complete + candidate answer. Non-sensitive context needed by the child must be included in its invocation or + fetched again; the child never sees parent history automatically. +- Do not return the candidate until an authorized review succeeds. On rejection, return only a + non-sensitive status. + +Important limitations: + +1. A normal subagent call is a background task. The parent can continue, post an interim answer, or + mishandle a later result unless its tools and instructions enforce the protocol. A waiting + workflow tool that calls the child provides a more reliable orchestration boundary. +2. The built-in root-copy `agent` shares root tools and sandbox. A declared specialist is the + narrower isolation boundary for this use case. +3. Child HITL is proxied to the parent's channel. Embedding the sensitive draft in a standard + question therefore puts it in the parent `input.requested` event and default Slack thread post. +4. The parent sees any returned candidate as a tool/task result before its next model call. If the + parent model is expected to render or rewrite it, the reviewed bytes are no longer the delivered + bytes. +5. Denial or task cancellation does not erase the child session. Parent-session finalization cleans + up live child ownership, but eve does not promise physical deletion of child history, workflow + logs, sandbox data, traces, or external tool records. +6. Declared subagents have isolated authored slots. Shared tools, connections, skills, and sandbox + policy must be deliberately mounted or authored in the child. +7. Sensitive work is invisible to parent hooks but not automatically invisible to child hooks, + child streams, logs, instrumentation, provider retention, or the shared sandbox of a root copy. + +**Use when:** an application can accept a protected child session as the sensitive persistence +boundary and can provide its own private review UI. + +**Limitation:** it is isolation, not transactional commit or secure erase. + +### Option C: a hook after tools and before Slack synthesis + +This does not fit the current hook model and has two timing problems. + +First, “after all tool calls” is not a stable model-loop boundary. A model may interleave prose and +tool calls, make more calls after seeing results, or finish directly. A pre-synthesis gate would +review permission to use data, not the actual answer. To vet the answer, the gate must be after +synthesis and before release. + +Second, by either point the current pipeline has crossed boundaries the proposal wants to guard: + +- prior step tool results are already in the durable working history used for synthesis; +- text may already have produced content events and telemetry; +- on `message.completed`, Slack's channel handler runs before authored hooks; +- hooks cannot replace, delay, or acknowledge events, and a throw cannot undo side effects or the + stream write. + +Overriding Slack's `message.completed` handler can suppress the final post, but it still leaves the +answer in model history and ordinary events. It also needs custom handling for deltas, reasoning, +actions, failures, retries, and eventual authenticated release. Wiping selected calls afterward is +both too late and incompatible with eve's append-only/provider-valid history assumptions. + +**Use when:** the requirement is only “do not post the final Slack message automatically,” and +persisted history/events are allowed. + +**Limitation:** it cannot provide the requested confidentiality or rollback semantics. + +## Other viable application-level designs + +### One confidential workflow with direct release + +The strongest design available without changing eve is: + +```text +root model + -> waiting workflow tool (receives only non-sensitive request/context) + -> declared confidential subagent fetches + synthesizes candidate + -> protected review store / private Slack or web UI + -> authorized approve, reject, or edit + -> workflow posts the exact approved text idempotently + <- { delivered: true } or { rejected: true } +``` + +The workflow should return only a non-sensitive receipt. The root should not receive or re-render the +candidate. Sensitive tools enforce resource authorization and exist only in the child. The direct +post needs an idempotency key derived from session, turn, and review request so workflow replay +cannot duplicate release. + +This design preserves the root context but accepts sensitive persistence in the child, workflow, +model provider, and review store. A bespoke review callback must authenticate the responder, bind +the decision to the candidate hash and intended Slack destination, expire it, and reject replay. +Use an authenticated web page or DM when Slack ephemeral interactive support is insufficient; do +not put the draft in a normal `ask.prompt`. + +### Separate private session or deployment + +Route the request to a private eve session or separate deployment whose channel, hooks, storage, +telemetry, model retention, and credentials are configured for sensitive data. After review, send +only the approved text to the public destination. This is operationally heavier but gives a clearer +security perimeter than relying on prompt discipline inside one root session. + +## Proposed framework direction: quarantined turns + +A first-class feature should make review a transaction over a candidate turn, not a callback over +already-public events. + +```text +canonical session checkpoint + | + v +private candidate turn -- tool calls/results --> terminal draft + | | + | sensitivity = join(labels) + | | + +---------------------- review requested privately + | + +---------------------+--------------------+ + | | + approve/edit reject + | | + commit approved projection + post exact bytes destroy candidate; + atomically (or outbox-idempotently) commit safe status only +``` + +### Authoring contract + +The runtime needs a trusted classifier rather than a convention the model interprets. A compact +surface could let tool definitions project provenance and sensitivity from their full result, while +a channel or agent review policy decides which labels require review. Classification is monotonic +for the candidate turn: once sensitive, later non-sensitive calls cannot clear it. + +The review policy should define: + +- whether a candidate requires review from its accumulated labels and provenance; +- who may view and answer it, evaluated again at response time; +- which private renderer or review destination receives it; +- retention and denial behavior; +- whether approval can be binary or may replace the draft with reviewed text. + +The model must not be able to set or downgrade these fields. Tool-level resource authorization +remains mandatory and separate. + +### Commit semantics + +- The canonical session commits the accepted user input, but not candidate assistant/tool messages + while the turn runs. +- Candidate snapshots are durable, encrypted or equivalently protected, access controlled to the + review policy, and retention bounded. They are not served by the ordinary session stream. +- Contentful `action.result`, reasoning, text delta, and completion events stay in a privileged + candidate stream. Ordinary channel handlers, hooks, memory, and telemetry receive either metadata + or a redacted projection until approval. +- At terminal synthesis, eve parks on a review request containing an opaque candidate id. The + private renderer resolves the candidate under authorization rather than copying the draft into a + public event payload. +- Approval commits a minimized canonical projection. Prefer the final approved answer plus + provenance/audit metadata, omitting the hidden tool-call branch entirely; this avoids retaining + raw source data and avoids invalid call/result pairs. Applications that truly need raw approved + history may opt into promoting the complete branch. +- Delivery uses the reviewed text directly, with no second model call. Commit and channel send use + an outbox/idempotency record so either can retry without duplicate or mismatched release. +- Rejection, expiry, cancellation, or reset destroys the candidate according to its retention + policy and appends at most a generic non-sensitive outcome to canonical history. +- A candidate hash binds the decision to the draft, provenance set, destination, and policy + version. Any change creates a new review request. + +### Why this belongs below hooks + +The harness owns model steps and canonical history; the execution layer owns durable snapshots and +turn parking; the channel adapter owns outward side effects. Only a boundary spanning those three +can ensure that a candidate is durable enough to resume while remaining absent from ordinary +history and delivery. General stream hooks should remain observation-oriented. + +## Recommendation + +1. **Ship serialized history restoration as the framework primitive.** Add + `session.restoreHistory({ to })` alongside `clear` and `compact`. Accept only an exact prefix of + the current history and reject an out-of-range index. +2. **Add the minimal pre-release capability needed by this use case.** Run + `beforeResponseRelease` after terminal synthesis while `message.completed` is held. Let the hook + call `candidate.history.restoreTo(index)`; restoration suppresses the pending response, while no + call releases it normally. +3. **Leave policy and coarse rollback scope to the application.** The hook decides where the + relevant request begins, how sensitivity is represented, and whether approval matches the + candidate. Restoring an earlier prefix may discard unrelated intervening input. +4. **Do not infer task or side-effect ownership from history.** Initial restoration changes future + model context only. It does not cancel work or retract events, memory writes, tool effects, + traces, or messages already released. +5. **Keep quarantined turns as a stronger follow-on.** Add private candidate storage and event + projection only for applications that must prevent unapproved content from reaching internal + consumers, rather than making that larger lifecycle part of the narrow restoration feature. +6. **Use tool approval and subagent isolation where their existing boundaries fit.** Retrieval + authorization remains separate from response review, and a confidential subagent remains an + application architecture rather than the history-restoration abstraction. + +## Validation requirements for a framework implementation + +- Unit: sensitivity joins cannot downgrade; candidate hashes bind content, provenance, destination, + and policy; response authorization fails closed; commit projection never leaves orphaned tool + calls. +- Integration: several mixed-sensitivity calls produce one review; no ordinary hook, memory writer, + channel handler, or content telemetry sees the candidate before approval; rejection and expiry + leave canonical history free of candidate content; edited approval posts and persists identical + bytes. +- Scenario: crash/redeploy at every tool, synthesis, review, commit, and delivery boundary resumes + without losing the candidate, leaking it, or posting twice; cancellation/reset clean up the + candidate; concurrent and replayed decisions settle once. +- E2E: a Slack fixture proves the preview is visible only to the authorized reviewer, an + unauthorized interaction cannot approve, denial posts nothing sensitive, and approval produces + exactly one reviewed message. + +## Sources inspected + +- `docs/tools/human-in-the-loop.md` +- `docs/tools/workflows.mdx` +- `docs/subagents/index.mdx` +- `docs/guides/hooks.md` +- `docs/channels/slack.mdx` +- `packages/eve/src/harness/tool-loop.ts` +- `packages/eve/src/harness/emission.ts` +- `packages/eve/src/harness/step-hooks.ts` +- `packages/eve/src/execution/workflow-steps.ts` +- `packages/eve/src/execution/turn-workflow.ts` +- `packages/eve/src/execution/durable-session-store.ts` +- `packages/eve/src/public/channels/slack/defaults.ts` +- `packages/eve/src/channel/adapter.ts` +- `packages/eve/src/context/hook-lifecycle.ts`