diff --git a/packages/eve/src/channel/routes.ts b/packages/eve/src/channel/routes.ts index d2f5f2bfc..cd67677ae 100644 --- a/packages/eve/src/channel/routes.ts +++ b/packages/eve/src/channel/routes.ts @@ -43,6 +43,7 @@ export interface RouteHandlerArgs { export interface SendPayload { readonly message?: string | UserContent; + readonly [key: string]: unknown; readonly inputResponses?: readonly InputResponse[]; /** * Context strings contributed by the channel. eve appends each entry diff --git a/packages/eve/src/channel/send.test.ts b/packages/eve/src/channel/send.test.ts index 17c078cb0..a17139fa9 100644 --- a/packages/eve/src/channel/send.test.ts +++ b/packages/eve/src/channel/send.test.ts @@ -4,12 +4,12 @@ import type { ChannelAdapter } from "#channel/adapter.js"; import { createSendFn } from "#channel/send.js"; import type { RunHandle, Runtime } from "#channel/types.js"; import { RuntimeNoActiveSessionError } from "#execution/runtime-errors.js"; -import type { MessageStreamEvent } from "#protocol/message.js"; +import type { HandleMessageStreamEvent } from "#protocol/message.js"; function createMockRunHandle(): RunHandle { return { continuationToken: "test:token", - events: new ReadableStream(), + events: new ReadableStream(), sessionId: "mock-session-id", }; } @@ -20,7 +20,7 @@ function createRuntime(deliverError: unknown): Runtime { deliver: vi.fn().mockRejectedValue(deliverError), resolveSession: vi.fn(), run: vi.fn().mockResolvedValue(createMockRunHandle()), - getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), + getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), getStreamTailIndex: vi.fn().mockResolvedValue(-1), terminateSession: vi.fn(), }; @@ -77,6 +77,31 @@ describe("createSendFn", () => { expect(runtime.run).not.toHaveBeenCalled(); }); + it("preserves channel-specific fields through delivery", async () => { + const runtime: Runtime = { + cancelTurn: vi.fn(), + deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }), + resolveSession: vi.fn(), + run: vi.fn(), + getEventStream: vi.fn(), + getStreamTailIndex: vi.fn(), + terminateSession: vi.fn(), + }; + const send = createSendFn(runtime, ADAPTER, "test"); + + await send( + { + inputResponses: [{ optionId: "approve", requestId: "approval-1" }], + pendingApprovalCards: { "approval-1": { messageTs: "123.456" } }, + }, + { auth: null, continuationToken: "token" }, + ); + + expect(vi.mocked(runtime.deliver).mock.calls[0]?.[0].payload).toMatchObject({ + pendingApprovalCards: { "approval-1": { messageTs: "123.456" } }, + }); + }); + it("forwards context through deliver and run payloads", async () => { const context = ["thread background"]; const deliverRuntime: Runtime = { @@ -84,7 +109,7 @@ describe("createSendFn", () => { deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }), resolveSession: vi.fn(), run: vi.fn().mockResolvedValue(createMockRunHandle()), - getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), + getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), getStreamTailIndex: vi.fn().mockResolvedValue(-1), terminateSession: vi.fn(), }; @@ -119,7 +144,7 @@ describe("createSendFn", () => { deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }), resolveSession: vi.fn(), run: vi.fn().mockResolvedValue(createMockRunHandle()), - getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), + getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), getStreamTailIndex: vi.fn().mockResolvedValue(-1), terminateSession: vi.fn(), }; @@ -149,7 +174,7 @@ describe("createSendFn", () => { deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }), resolveSession: vi.fn(), run: vi.fn().mockResolvedValue(createMockRunHandle()), - getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), + getEventStream: vi.fn().mockResolvedValue(new ReadableStream()), getStreamTailIndex: vi.fn().mockResolvedValue(-1), terminateSession: vi.fn(), }; diff --git a/packages/eve/src/channel/send.ts b/packages/eve/src/channel/send.ts index 2ed42b423..57653a923 100644 --- a/packages/eve/src/channel/send.ts +++ b/packages/eve/src/channel/send.ts @@ -31,6 +31,7 @@ export function createSendFn( inputResponses, context, outputSchema, + ...channelPayload } = normalizeSendInput(input); const message = serializeUrlFilePartsInMessage(rawMessage); @@ -39,7 +40,7 @@ export function createSendFn( auth, continuationToken, requestId: metadata.requestId, - payload: { inputResponses, message, context, outputSchema }, + payload: { ...channelPayload, inputResponses, message, context, outputSchema }, }; const { sessionId } = await runtime.deliver(deliverInput); diff --git a/packages/eve/src/client/message-reducer.test.ts b/packages/eve/src/client/message-reducer.test.ts index 0f393bb2e..1fc13ad3d 100644 --- a/packages/eve/src/client/message-reducer.test.ts +++ b/packages/eve/src/client/message-reducer.test.ts @@ -1,12 +1,12 @@ import { describe, expect, it } from "vitest"; import { defaultMessageReducer } from "#client/message-reducer.js"; -import { stampTestEvents } from "#internal/testing/events.js"; import { createActionResultEvent, createActionsRequestedEvent, createAuthorizationCompletedEvent, createAuthorizationRequiredEvent, + createApprovalSettledEvent, createInputRequestedEvent, createMessageAppendedEvent, createMessageCompletedEvent, @@ -15,21 +15,8 @@ import { createResultCompletedEvent, createStepStartedEvent, createTurnCancelledEvent, - type UnstampedMessageStreamEvent, } from "#protocol/message.js"; -function reduceServerEvents( - reducer: ReturnType, - data: ReturnType["initial"]>, - events: readonly UnstampedMessageStreamEvent[], -) { - let next = data; - for (const event of stampTestEvents(events)) { - next = reducer.reduce(next, event); - } - return next; -} - describe("defaultMessageReducer", () => { it("projects messages, reasoning, and actions into UIMessage-compatible parts", () => { const reducer = defaultMessageReducer(); @@ -43,13 +30,17 @@ describe("defaultMessageReducer", () => { }, type: "client.message.submitted", }); - data = reduceServerEvents(reducer, data, [ + data = reducer.reduce( + data, createReasoningCompletedEvent({ reasoning: "Need the weather tool.", sequence: 1, stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createActionsRequestedEvent({ actions: [ { @@ -63,6 +54,9 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createActionResultEvent({ result: { callId: "call_1", @@ -74,7 +68,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -123,7 +117,8 @@ describe("defaultMessageReducer", () => { it("projects an action result without a preceding action request", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + const data = reducer.reduce( + reducer.initial(), createActionResultEvent({ result: { callId: "call_1", @@ -135,7 +130,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -169,7 +164,8 @@ describe("defaultMessageReducer", () => { it("projects denied tool output distinctly from generic failures", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + const data = reducer.reduce( + reducer.initial(), createActionResultEvent({ result: { callId: "call_1", @@ -184,7 +180,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -224,14 +220,15 @@ describe("defaultMessageReducer", () => { const reducer = defaultMessageReducer(); let data = reducer.initial(); - data = reduceServerEvents(reducer, data, [ + data = reducer.reduce( + data, createResultCompletedEvent({ result: { title: "Done" }, sequence: 0, stepIndex: 0, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -249,7 +246,8 @@ describe("defaultMessageReducer", () => { it("projects authorization prompts into assistant message parts", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + const data = reducer.reduce( + reducer.initial(), createAuthorizationRequiredEvent({ authorization: { expiresAt: "2026-06-26T12:00:00.000Z", @@ -264,7 +262,7 @@ describe("defaultMessageReducer", () => { turnId: "turn_1", webhookUrl: "https://agent.example.com/eve/v1/connections/notion/callback/hook", }), - ]); + ); expect(data.messages).toEqual([ { @@ -298,7 +296,8 @@ describe("defaultMessageReducer", () => { it("updates the pending authorization part when authorization completes", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.reduce( + reducer.initial(), createAuthorizationRequiredEvent({ authorization: { displayName: "Notion", @@ -312,6 +311,10 @@ describe("defaultMessageReducer", () => { turnId: "turn_1", webhookUrl: "https://agent.example.com/eve/v1/connections/notion/callback/hook", }), + ); + + data = reducer.reduce( + data, createAuthorizationCompletedEvent({ authorization: { displayName: "Notion", @@ -323,7 +326,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_2", }), - ]); + ); expect(data.messages).toEqual([ { @@ -357,7 +360,8 @@ describe("defaultMessageReducer", () => { it("projects input requests onto tool approval parts", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + const data = reducer.reduce( + reducer.initial(), createInputRequestedEvent({ requests: [ { @@ -380,7 +384,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -424,9 +428,54 @@ describe("defaultMessageReducer", () => { ]); }); + it("marks shared approval UI only after durable settlement", () => { + const reducer = defaultMessageReducer(); + let data = reducer.reduce( + reducer.initial(), + createInputRequestedEvent({ + requests: [ + { + action: { callId: "call_1", input: {}, kind: "tool-call", toolName: "bash" }, + options: [ + { id: "approve", label: "Approve" }, + { id: "cancel", label: "Cancel" }, + ], + prompt: "Approve?", + requestId: "approval_1", + }, + ], + sequence: 0, + stepIndex: 0, + turnId: "turn_1", + }), + ); + + data = reducer.reduce( + data, + createApprovalSettledEvent({ + outcome: "approved", + requestId: "approval_1", + responderPrincipalId: "slack:T1:U1", + sequence: 1, + stepIndex: 0, + turnId: "turn_1", + }), + ); + + expect(data.messages[0]?.parts).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + approval: { approved: true, id: "approval_1", reason: undefined }, + state: "approval-responded", + }), + ]), + ); + }); + it("marks input requests as responded when the client submits a response", () => { const reducer = defaultMessageReducer(); - let data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.reduce( + reducer.initial(), createInputRequestedEvent({ requests: [ { @@ -449,7 +498,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); data = reducer.reduce(data, { data: { @@ -504,7 +553,8 @@ describe("defaultMessageReducer", () => { it("merges resumed approval results back into the requested tool part", () => { const reducer = defaultMessageReducer(); - let data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.reduce( + reducer.initial(), createInputRequestedEvent({ requests: [ { @@ -527,7 +577,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_0", }), - ]); + ); data = reducer.reduce(data, { data: { @@ -536,12 +586,16 @@ describe("defaultMessageReducer", () => { }, type: "client.input.responded", }); - data = reduceServerEvents(reducer, data, [ + data = reducer.reduce( + data, createStepStartedEvent({ sequence: 1, stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createActionResultEvent({ result: { callId: "call_1", @@ -553,7 +607,7 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - ]); + ); const toolParts = data.messages.flatMap((message) => message.parts.filter((part) => part.type === "dynamic-tool"), @@ -593,20 +647,26 @@ describe("defaultMessageReducer", () => { it("keeps text from separate steps as separate parts", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.initial(); + + data = reducer.reduce( + data, createMessageCompletedEvent({ message: "First step.", sequence: 0, stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createMessageCompletedEvent({ message: "Second step.", sequence: 1, stepIndex: 1, turnId: "turn_1", }), - ]); + ); expect(data.messages).toEqual([ { @@ -642,7 +702,10 @@ describe("defaultMessageReducer", () => { // text parts by stepIndex alone drops the first run and reorders the second // ahead of the tool call. const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.initial(); + + data = reducer.reduce( + data, createMessageAppendedEvent({ messageDelta: "Checking Vienna", messageSoFar: "Checking Vienna", @@ -650,6 +713,9 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_0", }), + ); + data = reducer.reduce( + data, createMessageCompletedEvent({ finishReason: "tool-calls", message: "Checking Vienna first.", @@ -657,6 +723,9 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_0", }), + ); + data = reducer.reduce( + data, createActionsRequestedEvent({ actions: [ { @@ -670,6 +739,9 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_0", }), + ); + data = reducer.reduce( + data, createMessageAppendedEvent({ messageDelta: "Now Berlin", messageSoFar: "Now Berlin", @@ -677,13 +749,16 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_0", }), + ); + data = reducer.reduce( + data, createMessageCompletedEvent({ message: "Now checking Berlin.", sequence: 4, stepIndex: 0, turnId: "turn_0", }), - ]); + ); const assistant = data.messages.find((message) => message.id === "turn_0:assistant"); expect( @@ -695,7 +770,8 @@ describe("defaultMessageReducer", () => { it("finalizes partial streamed message and reasoning when the turn is cancelled", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.reduce( + reducer.initial(), createReasoningAppendedEvent({ reasoningDelta: "Thinking", reasoningSoFar: "Thinking", @@ -703,6 +779,9 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createMessageAppendedEvent({ messageDelta: "Partial", messageSoFar: "Partial", @@ -710,8 +789,8 @@ describe("defaultMessageReducer", () => { stepIndex: 0, turnId: "turn_1", }), - createTurnCancelledEvent({ sequence: 2, turnId: "turn_1" }), - ]); + ); + data = reducer.reduce(data, createTurnCancelledEvent({ sequence: 2, turnId: "turn_1" })); expect(data.messages).toEqual([ { @@ -742,13 +821,17 @@ describe("defaultMessageReducer", () => { it("removes streamed text for a null message completion", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ + let data = reducer.reduce( + reducer.initial(), createMessageCompletedEvent({ message: "Earlier step.", sequence: 0, stepIndex: 0, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createMessageAppendedEvent({ messageDelta: "", messageSoFar: "", @@ -756,13 +839,16 @@ describe("defaultMessageReducer", () => { stepIndex: 1, turnId: "turn_1", }), + ); + data = reducer.reduce( + data, createMessageCompletedEvent({ message: null, sequence: 1, stepIndex: 1, turnId: "turn_1", }), - ]); + ); expect(data.messages[0]?.parts).toEqual([ { type: "step-start" }, @@ -778,26 +864,24 @@ describe("defaultMessageReducer", () => { it("projects structured file parts from message.received onto the user message", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ - { - data: { - message: "describe this\n[file: report.pdf (application/pdf)]", - parts: [ - { text: "describe this", type: "text" }, - { - filename: "report.pdf", - mediaType: "application/pdf", - size: 4, - type: "file", - url: "https://files.example.com/report.pdf", - }, - ], - sequence: 1, - turnId: "turn_1", - }, - type: "message.received", + const data = reducer.reduce(reducer.initial(), { + data: { + message: "describe this\n[file: report.pdf (application/pdf)]", + parts: [ + { text: "describe this", type: "text" }, + { + filename: "report.pdf", + mediaType: "application/pdf", + size: 4, + type: "file", + url: "https://files.example.com/report.pdf", + }, + ], + sequence: 1, + turnId: "turn_1", }, - ]); + type: "message.received", + }); const userMessage = data.messages.find((message) => message.role === "user"); expect(userMessage?.parts).toEqual([ @@ -814,12 +898,10 @@ describe("defaultMessageReducer", () => { it("falls back to a single text part when message.received omits parts", () => { const reducer = defaultMessageReducer(); - const data = reduceServerEvents(reducer, reducer.initial(), [ - { - data: { message: "hello there", sequence: 1, turnId: "turn_1" }, - type: "message.received", - }, - ]); + const data = reducer.reduce(reducer.initial(), { + data: { message: "hello there", sequence: 1, turnId: "turn_1" }, + type: "message.received", + }); const userMessage = data.messages.find((message) => message.role === "user"); expect(userMessage?.parts).toEqual([{ state: "done", text: "hello there", type: "text" }]); diff --git a/packages/eve/src/client/message-reducer.ts b/packages/eve/src/client/message-reducer.ts index a033a6085..b80d0628b 100644 --- a/packages/eve/src/client/message-reducer.ts +++ b/packages/eve/src/client/message-reducer.ts @@ -168,6 +168,42 @@ function reduceMessageData(data: EveMessageData, event: EveAgentReducerEvent): E return next; } + case "approval.candidate": + // Candidate progress is responder-specific. Applications can consume the + // raw stream event for private UI without changing the shared tool part. + return data; + + case "approval.settled": { + const existing = findToolPartByApprovalId(data, event.data.requestId); + if (existing === undefined) return data; + if (event.data.outcome === "approved") { + return updateToolPart(data, existing.toolCallId, { + approval: { approved: true, id: event.data.requestId, reason: undefined }, + input: existing.input, + state: "approval-responded", + stepIndex: existing.stepIndex, + toolCallId: existing.toolCallId, + toolMetadata: existing.toolMetadata, + toolName: existing.toolName, + type: "dynamic-tool", + }); + } + return updateToolPart(data, existing.toolCallId, { + approval: { + approved: false, + id: event.data.requestId, + reason: "Tool execution was cancelled.", + }, + input: existing.input, + state: "output-denied", + stepIndex: existing.stepIndex, + toolCallId: existing.toolCallId, + toolMetadata: existing.toolMetadata, + toolName: existing.toolName, + type: "dynamic-tool", + }); + } + case "action.result": { const descriptor = normalizeActionResult(event.data.result); const existing = findToolPart(data, event.data.result.callId); diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index a14c54eee..1bba1a804 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -19,7 +19,6 @@ import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js import { runStep } from "#context/run-step.js"; import { deserializeContext, serializeContext } from "#context/serialize.js"; import { getHarnessEmissionState } from "#harness/emission.js"; -import { preserveSerializedAgentTraceState } from "#harness/agent-trace-context-store.js"; import { isSessionLimitDecline, isTurnCancellation, @@ -44,9 +43,8 @@ import { createAuthorizationCompletedEvent, createSessionStartedEvent, encodeMessageStreamEvent, - type UnstampedMessageStreamEvent, - stampMessageStreamEvent, - type MessageStreamEvent, + type HandleMessageStreamEvent, + timestampHandleMessageStreamEvent, } from "#protocol/message.js"; import { CallbackBaseUrlKey, @@ -163,11 +161,19 @@ export async function turnStep(rawInput: TurnStepInput): Promise + | Array<{ + candidateId?: string; + name: string; + authorization: ConnectionAuthorizationChallenge; + }> | undefined; if (pendingAuth && input.input?.kind === "deliver") { const authResults: Array<{ name: string } & AuthorizationResult> = []; - const completed: Array<{ name: string; authorization: ConnectionAuthorizationChallenge }> = []; + const completed: Array<{ + candidateId?: string; + name: string; + authorization: ConnectionAuthorizationChallenge; + }> = []; const remainingPayloads: DeliverPayload[] = []; for (const payload of input.input.payloads) { const cb = payload["authorizationCallback"] as @@ -182,7 +188,11 @@ export async function turnStep(rawInput: TurnStepInput): Promise => { + const emit = async (event: HandleMessageStreamEvent): Promise => { const toEmit = await callAdapterEventHandler(adapter, event, adapterCtx); setChannelContext(ctx, { ...adapter, state: { ...adapterCtx.state } }); - const stamped = stampMessageStreamEvent(toEmit); - await writer.write(encodeMessageStreamEvent(stamped)); - return stamped; + await writer.write(encodeMessageStreamEvent(timestampHandleMessageStreamEvent(toEmit))); + return toEmit; }; const handleEvent = async ( - event: UnstampedMessageStreamEvent, + event: HandleMessageStreamEvent, messages?: readonly import("ai").ModelMessage[], ): Promise => { const emitted = await emit(event); @@ -357,10 +365,11 @@ export async function turnStep(rawInput: TurnStepInput): Promise { it("tracks authorization-required state and provider expiry", () => { const first = create({ candidateId: "candidate-1", principalId: "U1" }); const state = markApprovalCandidateAuthorizationRequired({ + authorizationName: "candidate-1:github", candidateId: "candidate-1", expiresAt: 500, provider: "GitHub", @@ -99,6 +100,7 @@ describe("approval candidate state", () => { }); expect(getApprovalAuditState(state).activeCandidates[0]).toMatchObject({ + authorizationName: "candidate-1:github", expiresAt: 500, provider: "GitHub", status: "authorization-required", diff --git a/packages/eve/src/harness/approval-candidates.ts b/packages/eve/src/harness/approval-candidates.ts index b754d0139..57494b3ba 100644 --- a/packages/eve/src/harness/approval-candidates.ts +++ b/packages/eve/src/harness/approval-candidates.ts @@ -19,6 +19,7 @@ export interface ApprovalCandidateAuditRecord { readonly status: ApprovalCandidateStatus; readonly createdAt: number; readonly completedAt?: number; + readonly eventEmitted?: boolean; readonly expiresAt?: number; readonly provider?: string; readonly runtimeRevision?: string; @@ -38,6 +39,7 @@ export interface ApprovalSettlementAuditRecord { readonly requestId: string; readonly settledAt: number; readonly candidateId?: string; + readonly eventEmitted?: boolean; } interface ActiveApprovalCandidate { @@ -45,7 +47,9 @@ interface ActiveApprovalCandidate { readonly requestId: string; readonly responder: ApprovalResponderIdentity; readonly status: "pending" | "authorization-required"; + readonly authorizationName?: string; readonly createdAt: number; + readonly pendingEventEmitted?: boolean; readonly expiresAt: number; readonly provider?: string; readonly runtimeRevision?: string; @@ -115,8 +119,62 @@ export function createApprovalCandidate(input: { }; } +/** Marks the pending candidate event as emitted. */ +export function markApprovalCandidatePendingEventEmitted(input: { + readonly candidateId: string; + readonly state: SessionStateMap | undefined; +}): SessionStateMap | undefined { + const approvalState = readApprovalState(input.state); + const candidate = approvalState.activeCandidates[input.candidateId]; + if (candidate === undefined || candidate.pendingEventEmitted === true) return input.state; + return writeApprovalState(input.state, { + ...approvalState, + activeCandidates: { + ...approvalState.activeCandidates, + [input.candidateId]: { ...candidate, pendingEventEmitted: true }, + }, + }); +} + +/** Marks a terminal candidate history event as emitted. */ +export function markApprovalCandidateHistoryEventEmitted(input: { + readonly candidateId: string; + readonly state: SessionStateMap | undefined; +}): SessionStateMap | undefined { + const approvalState = readApprovalState(input.state); + let changed = false; + const candidateHistory = approvalState.candidateHistory.map((candidate) => { + if (candidate.candidateId !== input.candidateId || candidate.eventEmitted === true) { + return candidate; + } + changed = true; + return { ...candidate, eventEmitted: true }; + }); + return changed + ? writeApprovalState(input.state, { ...approvalState, candidateHistory }) + : input.state; +} + +/** Marks a terminal settlement event as emitted. */ +export function markApprovalSettlementEventEmitted(input: { + readonly requestId: string; + readonly state: SessionStateMap | undefined; +}): SessionStateMap | undefined { + const approvalState = readApprovalState(input.state); + const settlement = approvalState.settlements[input.requestId]; + if (settlement === undefined || settlement.eventEmitted === true) return input.state; + return writeApprovalState(input.state, { + ...approvalState, + settlements: { + ...approvalState.settlements, + [input.requestId]: { ...settlement, eventEmitted: true }, + }, + }); +} + /** Marks a candidate as waiting on a private authorization challenge. */ export function markApprovalCandidateAuthorizationRequired(input: { + readonly authorizationName: string; readonly candidateId: string; readonly expiresAt?: number; readonly provider?: string; @@ -127,6 +185,7 @@ export function markApprovalCandidateAuthorizationRequired(input: { if (candidate === undefined) return input.state; const nextCandidate: ActiveApprovalCandidate = { ...candidate, + authorizationName: input.authorizationName, expiresAt: input.expiresAt ?? candidate.expiresAt, provider: input.provider, status: "authorization-required", diff --git a/packages/eve/src/harness/authorization.test.ts b/packages/eve/src/harness/authorization.test.ts new file mode 100644 index 000000000..d6e456f81 --- /dev/null +++ b/packages/eve/src/harness/authorization.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; + +import { + getPendingAuthorization, + setPendingAuthorization, + type AuthorizationChallenge, +} from "#harness/authorization.js"; + +function challenge(name: string, candidateId: string): AuthorizationChallenge { + return { + candidateId, + challenge: { url: `https://example.com/${name}` }, + hookUrl: `https://eve.example/${name}`, + name, + }; +} + +describe("pending authorization state", () => { + it("merges concurrent candidate challenges by authorization name", () => { + const first = setPendingAuthorization(undefined, { + challenges: [challenge("candidate-1:github", "candidate-1")], + }); + const second = setPendingAuthorization(first, { + challenges: [challenge("candidate-2:github", "candidate-2")], + }); + + expect(getPendingAuthorization(second)?.challenges).toEqual([ + expect.objectContaining({ candidateId: "candidate-1", name: "candidate-1:github" }), + expect.objectContaining({ candidateId: "candidate-2", name: "candidate-2:github" }), + ]); + }); + + it("replaces a repeated challenge without duplicating it", () => { + const first = setPendingAuthorization(undefined, { + challenges: [challenge("candidate-1:github", "candidate-1")], + }); + const second = setPendingAuthorization(first, { + challenges: [ + { + ...challenge("candidate-1:github", "candidate-1"), + hookUrl: "https://eve.example/refreshed", + }, + ], + }); + + expect(getPendingAuthorization(second)?.challenges).toEqual([ + expect.objectContaining({ hookUrl: "https://eve.example/refreshed" }), + ]); + }); +}); diff --git a/packages/eve/src/harness/authorization.ts b/packages/eve/src/harness/authorization.ts index 76bdd41f9..b7d9316c6 100644 --- a/packages/eve/src/harness/authorization.ts +++ b/packages/eve/src/harness/authorization.ts @@ -46,6 +46,7 @@ const AUTHORIZATION_PENDING_BRAND = "__eveAuthorizationPending" as const; // --------------------------------------------------------------------------- export interface AuthorizationChallenge { + readonly candidateId?: string; readonly name: string; readonly challenge: ConnectionAuthorizationChallenge; readonly hookUrl: string; @@ -104,6 +105,7 @@ export function requestAuthorization( export function redactSignalResume(signal: AuthorizationSignal): AuthorizationSignal { return requestAuthorization( signal.challenges.map((entry) => ({ + candidateId: entry.candidateId, name: entry.name, challenge: entry.challenge, hookUrl: entry.hookUrl, @@ -240,7 +242,13 @@ export function setPendingAuthorization( sessionState: Record | undefined, value: PendingAuthorizationState, ): Record { - return { ...sessionState, [PENDING_AUTHORIZATION_KEY]: value }; + const existing = getPendingAuthorization(sessionState)?.challenges ?? []; + const byName = new Map(existing.map((challenge) => [challenge.name, challenge])); + for (const challenge of value.challenges) byName.set(challenge.name, challenge); + return { + ...sessionState, + [PENDING_AUTHORIZATION_KEY]: { challenges: [...byName.values()] }, + }; } export function clearPendingAuthorization( diff --git a/packages/eve/src/harness/authorize-approval-response.ts b/packages/eve/src/harness/authorize-approval-response.ts index 6e915f384..e7274e049 100644 --- a/packages/eve/src/harness/authorize-approval-response.ts +++ b/packages/eve/src/harness/authorize-approval-response.ts @@ -1,5 +1,7 @@ import type { SessionAuthContext } from "#channel/types.js"; import { buildCallbackContext } from "#context/build-callback-context.js"; +import { contextStorage } from "#context/container.js"; +import { AuthKey } from "#context/keys.js"; import { buildApprovalResponseAuth, handleApprovalResponseAuthorizationError, @@ -7,6 +9,7 @@ import { import { cancelApprovalRequest, createApprovalCandidate, + expireApprovalCandidates, finishApprovalCandidate, getApprovalAuditState, markApprovalCandidateAuthorizationRequired, @@ -30,6 +33,12 @@ const APPROVAL_CANDIDATE_TTL_MS = 10 * 60_000; export type PendingApprovalAuthorizationResult = | { readonly kind: "continue"; readonly session: HarnessSession; readonly stepInput?: StepInput } + | { + readonly candidateId: string; + readonly kind: "candidate-created"; + readonly requestId: string; + readonly session: HarnessSession; + } | { readonly authorization: AuthorizationSignal; readonly candidateId: string; @@ -54,13 +63,31 @@ export async function authorizePendingApprovalResponse(input: { readonly stepInput?: StepInput; readonly tools: HarnessToolMap; }): Promise { + const now = input.now ?? Date.now(); + input = { + ...input, + session: { + ...input.session, + state: expireApprovalCandidates({ now, state: input.session.state }), + }, + }; const batch = getPendingInputBatch(input.session.state); + const explicitResponse = input.stepInput?.inputResponses?.find((entry) => + batch?.requests.some( + (request) => request.requestId === entry.requestId && isApprovalRequest(request), + ), + ); const activeCandidate = getApprovalAuditState(input.session.state).activeCandidates.find( (candidate) => - candidate.status === "authorization-required" && - getAuthorizationResult(candidate.candidateId) !== undefined, + (explicitResponse === undefined && + candidate.status === "pending" && + candidate.pendingEventEmitted === true) || + (candidate.status === "authorization-required" && + candidate.authorizationName !== undefined && + getAuthorizationResult(candidate.authorizationName) !== undefined), ); const response = + explicitResponse ?? input.stepInput?.inputResponses?.find((entry) => batch?.requests.some( (request) => request.requestId === entry.requestId && isApprovalRequest(request), @@ -87,7 +114,7 @@ export async function authorizePendingApprovalResponse(input: { const settled = cancelApprovalRequest({ actor: responder, requestId: response.requestId, - settledAt: input.now ?? Date.now(), + settledAt: now, state: input.session.state, }); return { @@ -109,7 +136,10 @@ export async function authorizePendingApprovalResponse(input: { return { kind: "continue", session: input.session, stepInput: input.stepInput }; } - const responder = buildCallbackContext().session.auth.current; + const responder = + activeCandidate === undefined + ? buildCallbackContext().session.auth.current + : responderFromCandidate(activeCandidate.responder); if (responder === null) { return rejectWithoutCandidate( input, @@ -117,9 +147,11 @@ export async function authorizePendingApprovalResponse(input: { "An authenticated responder is required.", ); } - const now = input.now ?? Date.now(); const candidateId = activeCandidate?.candidateId ?? approvalCandidateId(request.requestId, responder); + if (activeCandidate !== undefined) { + contextStorage.getStore()?.setVirtualContext(AuthKey, responder); + } const created = createApprovalCandidate({ candidateId, createdAt: now, @@ -130,6 +162,14 @@ export async function authorizePendingApprovalResponse(input: { state: input.session.state, }); let session = { ...input.session, state: created.state }; + if (created.result.kind === "created") { + return { + candidateId, + kind: "candidate-created", + requestId: response.requestId, + session, + }; + } if ( activeCandidate === undefined && (created.result.kind === "duplicate" || created.result.kind === "stale") @@ -221,6 +261,7 @@ export async function authorizePendingApprovalResponse(input: { session = { ...session, state: markApprovalCandidateAuthorizationRequired({ + authorizationName: authorization.challenges[0]?.name ?? candidateId, candidateId, expiresAt: providerExpiresAt, provider: authorization.challenges[0]?.challenge.displayName, @@ -307,6 +348,12 @@ function removeInputResponse( }; } +function responderFromCandidate( + responder: Pick, +): SessionAuthContext { + return { ...responder, attributes: {} }; +} + function approvalCandidateId(requestId: string, responder: SessionAuthContext): string { const principal = [ responder.authenticator, diff --git a/packages/eve/src/harness/input-requests.test.ts b/packages/eve/src/harness/input-requests.test.ts index 3181766d7..e6a4ec5c7 100644 --- a/packages/eve/src/harness/input-requests.test.ts +++ b/packages/eve/src/harness/input-requests.test.ts @@ -2,7 +2,8 @@ import { jsonSchema, type ModelMessage } from "ai"; import { describe, expect, it } from "vitest"; import { ContextContainer, contextStorage } from "#context/container.js"; -import { SessionKey } from "#context/keys.js"; +import { AuthKey, SessionKey } from "#context/keys.js"; +import { PendingAuthorizationResultKey } from "#harness/authorization.js"; import { once } from "#public/tools/approval/approval-helpers.js"; import type { InputRequest } from "#runtime/input/types.js"; import type { HarnessToolDefinition } from "#harness/execute-tool.js"; @@ -911,8 +912,37 @@ describe("authorizePendingApprovalResponse", () => { return contextStorage.run(ctx, run); } + async function continueCreatedCandidate( + result: Awaited>, + tools: HarnessToolMap, + ) { + if (result.kind !== "candidate-created") return result; + const state = result.session.state?.["eve.runtime.hitl.approvalState"] as { + activeCandidates: Record>; + }; + const activeCandidates = Object.fromEntries( + Object.entries(state.activeCandidates).map(([id, candidate]) => [ + id, + { ...candidate, pendingEventEmitted: true }, + ]), + ); + return await approvalContext(() => + authorizePendingApprovalResponse({ + now: 101, + session: { + ...result.session, + state: { + ...result.session.state, + "eve.runtime.hitl.approvalState": { ...state, activeCandidates }, + }, + }, + tools, + }), + ); + } + it("fails closed when a required authorizer is missing", async () => { - const result = await approvalContext(() => + let result = await approvalContext(() => authorizePendingApprovalResponse({ now: 100, session: pendingSession(), @@ -922,11 +952,11 @@ describe("authorizePendingApprovalResponse", () => { tools: new Map(), }), ); + result = await continueCreatedCandidate(result, new Map()); expect(result).toMatchObject({ kind: "failed", safeReason: "Approval authorization is temporarily unavailable. Please try again.", - stepInput: { inputResponses: [] }, }); expect(resolvePendingInput({ session: result.session }).outcome).toBe("unresolved"); }); @@ -949,7 +979,7 @@ describe("authorizePendingApprovalResponse", () => { }, ], ]); - const result = await approvalContext(() => + let result = await approvalContext(() => authorizePendingApprovalResponse({ now: 100, session: pendingSession(), @@ -959,11 +989,11 @@ describe("authorizePendingApprovalResponse", () => { tools, }), ); + result = await continueCreatedCandidate(result, tools); expect(result).toMatchObject({ kind: "rejected", safeReason: "U1 lacks repository write access.", - stepInput: { inputResponses: [] }, }); expect(resolvePendingInput({ session: result.session }).outcome).toBe("unresolved"); }); @@ -983,7 +1013,7 @@ describe("authorizePendingApprovalResponse", () => { }, ], ]); - const result = await approvalContext(() => + let result = await approvalContext(() => authorizePendingApprovalResponse({ now: 100, session: pendingSession(), @@ -993,6 +1023,7 @@ describe("authorizePendingApprovalResponse", () => { tools, }), ); + result = await continueCreatedCandidate(result, tools); expect(result.kind).toBe("continue"); if (result.kind !== "continue") throw new Error("Expected allowed candidate."); @@ -1057,7 +1088,7 @@ describe("authorizePendingApprovalResponse", () => { }, ], ]); - const result = await approvalContext(() => + let result = await approvalContext(() => authorizePendingApprovalResponse({ now: 100, session, @@ -1070,6 +1101,7 @@ describe("authorizePendingApprovalResponse", () => { tools, }), ); + result = await continueCreatedCandidate(result, tools); expect(result.kind).toBe("continue"); if (result.kind !== "continue") throw new Error("Expected first candidate to settle."); @@ -1081,6 +1113,145 @@ describe("authorizePendingApprovalResponse", () => { ).toBe("unresolved"); }); + it("starts a second responder candidate instead of resuming another user's pending candidate", async () => { + const base = pendingSession(); + const session = { + ...base, + state: { + ...base.state, + "eve.runtime.hitl.approvalState": { + activeCandidates: { + "candidate-original": { + candidateId: "candidate-original", + createdAt: 100, + expiresAt: 700, + pendingEventEmitted: true, + requestId: "approval-1", + responder: { + authenticator: "slack-webhook", + issuer: "slack:T1", + principalId: "U_ORIGINAL", + principalType: "user", + }, + status: "pending", + }, + }, + candidateHistory: [], + settlements: {}, + }, + }, + }; + const tools: HarnessToolMap = new Map([ + [ + "create_issue", + { + approval: { authorizeResponse: () => "allowed", policy: () => "user-approval" }, + description: "Create issue", + inputSchema: jsonSchema({ type: "object" }), + name: "create_issue", + }, + ], + ]); + + const result = await approvalContext(() => + authorizePendingApprovalResponse({ + now: 200, + session, + stepInput: { inputResponses: [{ optionId: "approve", requestId: "approval-1" }] }, + tools, + }), + ); + + expect(result).toMatchObject({ kind: "candidate-created" }); + const active = ( + result.session.state?.["eve.runtime.hitl.approvalState"] as { + activeCandidates: Record; + } + ).activeCandidates; + expect(Object.values(active).map((candidate) => candidate.responder.principalId)).toEqual([ + "U_ORIGINAL", + "U1", + ]); + }); + + it("restores the persisted candidate responder before resumed policy runs", async () => { + const candidateState = { + "eve.runtime.hitl.approvalState": { + activeCandidates: { + "candidate-1": { + authorizationName: "candidate-1:github", + candidateId: "candidate-1", + createdAt: 100, + expiresAt: 700, + requestId: "approval-1", + responder: { + authenticator: "slack-webhook", + issuer: "slack:T1", + principalId: "U_ORIGINAL", + principalType: "user", + }, + status: "authorization-required", + }, + }, + candidateHistory: [], + settlements: {}, + }, + }; + const session = { + ...pendingSession(), + state: { ...pendingSession().state, ...candidateState }, + }; + const tools: HarnessToolMap = new Map([ + [ + "create_issue", + { + approval: { + authorizeResponse: ({ responder }) => + responder.principalId === "U_ORIGINAL" + ? "allowed" + : { safeReason: "wrong responder", status: "rejected" }, + policy: () => "user-approval", + }, + description: "Create issue", + inputSchema: jsonSchema({ type: "object" }), + name: "create_issue", + }, + ], + ]); + const ctx = new ContextContainer(); + ctx.set(AuthKey, { + attributes: {}, + authenticator: "slack-webhook", + issuer: "slack:T1", + principalId: "U_LATEST", + principalType: "user", + }); + ctx.set(PendingAuthorizationResultKey, [ + { + callback: { method: "GET", params: { code: "test" } }, + hookUrl: "https://eve.example/callback", + name: "candidate-1:github", + }, + ]); + ctx.set(SessionKey, { + auth: { current: ctx.require(AuthKey), initiator: null }, + sessionId: "sess-test", + turn: { id: "turn-test", sequence: 1 }, + }); + + const result = await contextStorage.run(ctx, () => + authorizePendingApprovalResponse({ + now: 200, + session, + stepInput: { inputResponses: [{ optionId: "approve", requestId: "approval-1" }] }, + tools, + }), + ); + + expect(result.kind).toBe("continue"); + expect(ctx.get(AuthKey)?.principalId).toBe("U_ORIGINAL"); + }); + it("cancels without running the response authorizer", async () => { const tools: HarnessToolMap = new Map([ [ diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index 673484492..952a280d2 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -104,7 +104,12 @@ import { } from "#harness/input-extraction.js"; import { createToolResultMessagePartFromToolError } from "#harness/action-result-helpers.js"; import { buildTelemetryRuntimeContext } from "#harness/instrumentation-runtime-context.js"; -import { getApprovalAuditState } from "#harness/approval-candidates.js"; +import { + getApprovalAuditState, + markApprovalCandidateHistoryEventEmitted, + markApprovalCandidatePendingEventEmitted, + markApprovalSettlementEventEmitted, +} from "#harness/approval-candidates.js"; import { authorizePendingApprovalResponse } from "#harness/authorize-approval-response.js"; import { consumeDeferredStepInput, @@ -574,18 +579,67 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { tools: responseAuthorizationTools, }); session = authorized.session; - if (emit && authorized.kind !== "continue" && authorized.kind !== "duplicate") { + if (authorized.kind === "candidate-created") { + return { next: runStep, session }; + } + + const pendingCandidateEvent = getApprovalAuditState(session.state).activeCandidates.find( + (candidate) => candidate.pendingEventEmitted !== true, + ); + if (pendingCandidateEvent !== undefined) { + if (emit) { + await emit( + createApprovalCandidateEvent({ + candidateId: pendingCandidateEvent.candidateId, + outcome: "pending", + requestId: pendingCandidateEvent.requestId, + responderPrincipalId: pendingCandidateEvent.responder.principalId, + sequence: emissionState.sequence, + stepIndex: emissionState.stepIndex, + turnId: emissionState.turnId, + }), + ); + } + session = { + ...session, + state: markApprovalCandidatePendingEventEmitted({ + candidateId: pendingCandidateEvent.candidateId, + state: session.state, + }), + }; + return { next: runStep, session }; + } + + const pendingCandidateHistoryEvent = getApprovalAuditState(session.state).candidateHistory.find( + (candidate) => + candidate.eventEmitted !== true && + candidate.status !== "allowed" && + candidate.status !== "authorization-required", + ); + if (emit && pendingCandidateHistoryEvent !== undefined) { await emit( createApprovalCandidateEvent({ - candidateId: authorized.candidateId ?? `stale:${authorized.requestId}`, - outcome: authorized.kind === "authorization-required" ? "pending" : authorized.kind, - requestId: authorized.requestId, - safeReason: "safeReason" in authorized ? authorized.safeReason : undefined, + candidateId: pendingCandidateHistoryEvent.candidateId, + outcome: pendingCandidateHistoryEvent.status as Exclude< + typeof pendingCandidateHistoryEvent.status, + "allowed" | "authorization-required" + >, + requestId: pendingCandidateHistoryEvent.requestId, + responderPrincipalId: pendingCandidateHistoryEvent.responder.principalId, + safeReason: pendingCandidateHistoryEvent.safeReason, sequence: emissionState.sequence, stepIndex: emissionState.stepIndex, turnId: emissionState.turnId, }), ); + session = { + ...session, + state: markApprovalCandidateHistoryEventEmitted({ + candidateId: pendingCandidateHistoryEvent.candidateId, + state: session.state, + }), + }; + return { next: runStep, session }; } if (authorized.kind === "authorization-required") { const { challenges } = authorized.authorization; @@ -615,27 +669,28 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { }; } - if (emit && authorized.kind === "continue") { - const response = authorized.stepInput?.inputResponses?.find( - (entry) => entry.optionId === "approve" || entry.optionId === "cancel", + const pendingSettlementEvent = getApprovalAuditState(session.state).settlements.find( + (settlement) => settlement.eventEmitted !== true, + ); + if (emit && pendingSettlementEvent !== undefined) { + await emit( + createApprovalSettledEvent({ + outcome: pendingSettlementEvent.outcome === "allowed" ? "approved" : "cancelled", + requestId: pendingSettlementEvent.requestId, + responderPrincipalId: pendingSettlementEvent.actor.principalId, + sequence: emissionState.sequence, + stepIndex: emissionState.stepIndex, + turnId: emissionState.turnId, + }), ); - if (response !== undefined) { - const settlement = getApprovalAuditState(session.state).settlements.find( - (entry) => entry.requestId === response.requestId, - ); - if (settlement !== undefined) { - await emit( - createApprovalSettledEvent({ - outcome: settlement.outcome === "allowed" ? "approved" : "cancelled", - requestId: settlement.requestId, - responderPrincipalId: settlement.actor.principalId, - sequence: emissionState.sequence, - stepIndex: emissionState.stepIndex, - turnId: emissionState.turnId, - }), - ); - } - } + session = { + ...session, + state: markApprovalSettlementEventEmitted({ + requestId: pendingSettlementEvent.requestId, + state: session.state, + }), + }; + return { next: runStep, session }; } const pending = resolvePendingInput({ diff --git a/packages/eve/src/protocol/message.ts b/packages/eve/src/protocol/message.ts index 97b361c0e..5dc9bb5c0 100644 --- a/packages/eve/src/protocol/message.ts +++ b/packages/eve/src/protocol/message.ts @@ -6,7 +6,6 @@ import { isSerializedUrlFilePart, } from "#internal/attachments/url-refs.js"; import { decodeSandboxRef, isSandboxRefUrl } from "#internal/attachments/sandbox-refs.js"; -import { createEventId } from "#protocol/event-id.js"; import type { ConnectionAuthorizationChallenge } from "#public/connections/errors.js"; import type { RuntimeActionRequest, RuntimeActionResult } from "#runtime/actions/types.js"; import type { InputRequest, InputResponse } from "#runtime/input/types.js"; @@ -19,7 +18,7 @@ export const EVE_STREAM_TAIL_INDEX_HEADER = "x-eve-stream-tail-index"; export const EVE_STREAM_VERSION_HEADER = "x-eve-stream-version"; export const EVE_MESSAGE_STREAM_CONTENT_TYPE = "application/x-ndjson; charset=utf-8"; export const EVE_MESSAGE_STREAM_FORMAT = "ndjson"; -export const EVE_MESSAGE_STREAM_VERSION = "20"; +export const EVE_MESSAGE_STREAM_VERSION = "19"; /** * eve-owned finish reason for one completed assistant step. @@ -48,21 +47,11 @@ export interface StepCompletedProviderMetadata { /** * Durable metadata attached to one persisted session stream event. * - * Stamped once, immediately before the event is written to the workflow-owned - * stream, and stored with it. Re-reading the stream — reconnecting, rewinding, - * or replaying a finished session — yields the same values every time. + * Runtime code stamps this immediately before writing the event to the + * workflow-owned stream so replay preserves the original timing. */ -export interface MessageStreamEventMeta { - /** ISO-8601 emission time. */ +export interface HandleMessageStreamEventMeta { readonly at: string; - /** - * Unique, lexicographically sortable identifier for this event. - * - * Stamped from stream version 20 on: rewinding into a session that started - * before the upgrade yields events whose `id` is absent despite this type, - * and those cannot be deduplicated. - */ - readonly id: string; } /** @@ -224,6 +213,36 @@ export interface ActionsRequestedStreamEvent { type: "actions.requested"; } +export type ApprovalCandidateOutcome = "pending" | "rejected" | "failed" | "timed-out" | "stale"; + +/** Safe lifecycle event for one responder-bound approval candidate. */ +export interface ApprovalCandidateStreamEvent { + data: { + candidateId: string; + outcome: ApprovalCandidateOutcome; + requestId: string; + responderPrincipalId: string; + safeReason?: string; + sequence: number; + stepIndex: number; + turnId: string; + }; + type: "approval.candidate"; +} + +/** Terminal durable settlement for one approval request. */ +export interface ApprovalSettledStreamEvent { + data: { + outcome: "approved" | "cancelled"; + requestId: string; + responderPrincipalId: string; + sequence: number; + stepIndex: number; + turnId: string; + }; + type: "approval.settled"; +} + /** * Stream event emitted when the harness needs human input before it can * continue the run. @@ -293,7 +312,7 @@ export interface SubagentStartedStreamEvent { export interface SubagentChildEventStreamEvent { data: { callId: string; - event: UnstampedMessageStreamEvent; + event: HandleMessageStreamEvent; subagentName: string; }; type: "subagent.event"; @@ -511,6 +530,7 @@ export interface CompactionCompletedStreamEvent { export interface AuthorizationRequiredStreamEvent { data: { authorization?: ConnectionAuthorizationChallenge; + candidateId?: string; description: string; name: string; sequence: number; @@ -542,6 +562,7 @@ export type ConnectionAuthorizationOutcome = AuthorizationOutcome; */ export interface AuthorizationCompletedStreamEvent { data: { + candidateId?: string; /** * The challenge from the matching `authorization.required` event, * journaled across the park. Lets channels keep rendering the @@ -592,12 +613,11 @@ export interface SessionCompletedStreamEvent { } /** - * Serializable event before eve stamps the durable stream envelope. - * - * Internal emitters use this type while constructing events. Public stream - * consumers receive {@link MessageStreamEvent}. + * Serializable stream event union for the durable message session flow. */ -export type UnstampedMessageStreamEvent = +export type HandleMessageStreamEvent = ( + | ApprovalCandidateStreamEvent + | ApprovalSettledStreamEvent | CompactionCompletedStreamEvent | CompactionRequestedStreamEvent | AuthorizationCompletedStreamEvent @@ -625,7 +645,10 @@ export type UnstampedMessageStreamEvent = | TurnCancelledStreamEvent | TurnCompletedStreamEvent | TurnFailedStreamEvent - | TurnStartedStreamEvent; + | TurnStartedStreamEvent +) & { + readonly meta?: HandleMessageStreamEventMeta; +}; /** * Stream events that represent an unrecovered turn/session failure. @@ -636,26 +659,22 @@ export type TurnFailureStreamEvent = | TurnFailedStreamEvent; /** - * One event read from an eve session stream. + * One public session stream event after runtime metadata has been stamped. * - * eve stamps the durable identity and emission time before writing the event. + * Runtime/execution code owns this stamping boundary. Replays must preserve the + * original `meta.at` value instead of recomputing it. */ -export type MessageStreamEvent = UnstampedMessageStreamEvent & { - readonly meta: MessageStreamEventMeta; +export type TimedHandleMessageStreamEvent = HandleMessageStreamEvent & { + readonly meta: HandleMessageStreamEventMeta; }; -/** - * @deprecated Use {@link MessageStreamEvent}. - */ -export type HandleMessageStreamEvent = MessageStreamEvent; - const textEncoder = new TextEncoder(); /** * Returns true when the current stream has reached a turn boundary or terminal * session outcome. */ -export function isCurrentTurnBoundaryEvent(event: UnstampedMessageStreamEvent): boolean { +export function isCurrentTurnBoundaryEvent(event: HandleMessageStreamEvent): boolean { return ( event.type === "session.completed" || event.type === "session.failed" || @@ -665,13 +684,10 @@ export function isCurrentTurnBoundaryEvent(event: UnstampedMessageStreamEvent): /** * Narrows a stream event to the failure events that terminate or poison a turn. - * - * Generic so narrowing keeps the input's stamping: a - * {@link MessageStreamEvent} narrows to a stamped failure event. */ -export function isTurnFailureEvent( - event: TEvent, -): event is TEvent & TurnFailureStreamEvent { +export function isTurnFailureEvent( + event: HandleMessageStreamEvent, +): event is TurnFailureStreamEvent { return ( event.type === "session.failed" || event.type === "step.failed" || event.type === "turn.failed" ); @@ -957,6 +973,7 @@ export function createActionsRequestedEvent(input: { */ export function createAuthorizationRequiredEvent(input: { readonly authorization?: ConnectionAuthorizationChallenge; + readonly candidateId?: string; readonly description: string; readonly name: string; readonly sequence: number; @@ -974,6 +991,9 @@ export function createAuthorizationRequiredEvent(input: { if (input.authorization !== undefined) { data.authorization = input.authorization; } + if (input.candidateId !== undefined) { + data.candidateId = input.candidateId; + } if (input.webhookUrl !== undefined) { data.webhookUrl = input.webhookUrl; } @@ -990,6 +1010,7 @@ export function createAuthorizationRequiredEvent(input: { */ export function createAuthorizationCompletedEvent(input: { readonly authorization?: ConnectionAuthorizationChallenge; + readonly candidateId?: string; readonly name: string; readonly outcome: AuthorizationOutcome; readonly reason?: string; @@ -1007,6 +1028,9 @@ export function createAuthorizationCompletedEvent(input: { if (input.authorization !== undefined) { data.authorization = input.authorization; } + if (input.candidateId !== undefined) { + data.candidateId = input.candidateId; + } if (input.reason !== undefined) { data.reason = input.reason; } @@ -1016,6 +1040,20 @@ export function createAuthorizationCompletedEvent(input: { }; } +/** Creates a safe candidate lifecycle event. */ +export function createApprovalCandidateEvent( + input: ApprovalCandidateStreamEvent["data"], +): ApprovalCandidateStreamEvent { + return { data: input, type: "approval.candidate" }; +} + +/** Creates a terminal approval settlement event. */ +export function createApprovalSettledEvent( + input: ApprovalSettledStreamEvent["data"], +): ApprovalSettledStreamEvent { + return { data: input, type: "approval.settled" }; +} + /** * Creates the `input.requested` event for one pending HITL batch. */ @@ -1423,18 +1461,21 @@ export function createSessionCompletedEvent(): SessionCompletedStreamEvent { } /** - * Stamps one session event with its durable identity and emission time. + * Stamps one session event with durable timing metadata immediately before it + * is written to the workflow-owned stream. * - * Runtime/execution code only, once per event, immediately before the write. - * One stamping seam is what makes the persisted stream and authored hooks - * observe the same `meta.id`. - */ -export function stampMessageStreamEvent(event: UnstampedMessageStreamEvent): MessageStreamEvent { + * Only runtime/execution code should call this. Keeping one stamping seam + * ensures every persisted event shares the same clock contract and replay never + * invents new timestamps. + */ +export function timestampHandleMessageStreamEvent( + event: HandleMessageStreamEvent, + at = new Date().toISOString(), +): TimedHandleMessageStreamEvent { return { ...event, meta: { - at: new Date().toISOString(), - id: createEventId(), + at, }, }; } @@ -1442,7 +1483,7 @@ export function stampMessageStreamEvent(event: UnstampedMessageStreamEvent): Mes /** * Encodes one message stream event as newline-delimited JSON. */ -export function encodeMessageStreamEvent(event: MessageStreamEvent): Uint8Array { +export function encodeMessageStreamEvent(event: TimedHandleMessageStreamEvent): Uint8Array { return textEncoder.encode(`${JSON.stringify(event)}\n`); } diff --git a/packages/eve/src/public/channels/slack/defaults.test.ts b/packages/eve/src/public/channels/slack/defaults.test.ts index 804814b09..12a500fd2 100644 --- a/packages/eve/src/public/channels/slack/defaults.test.ts +++ b/packages/eve/src/public/channels/slack/defaults.test.ts @@ -51,6 +51,102 @@ function authRequiredEvent( }; } +describe("defaultEvents approval lifecycle", () => { + it("sends candidate progress privately", async () => { + const { channel, postEphemeral } = buildChannelStub(); + const ctx = sessionContext({ + attributes: { user_id: "U777" }, + authenticator: "slack-webhook", + principalId: "slack:T1:U777", + principalType: "user", + }); + + await defaultEvents["approval.candidate"]!( + { + candidateId: "candidate-1", + outcome: "pending", + requestId: "approval-1", + responderPrincipalId: "slack:T1:U777", + sequence: 1, + stepIndex: 0, + turnId: "turn-1", + }, + channel, + ctx, + ); + + expect(postEphemeral).toHaveBeenCalledWith( + "U777", + "Checking whether you can approve this action…", + ); + }); + + it("delivers an immediate rejection to the current responder without prior mapping", async () => { + const { channel, postEphemeral } = buildChannelStub(); + const ctx = sessionContext({ + attributes: { user_id: "U777" }, + authenticator: "slack-webhook", + principalId: "slack:T1:U777", + principalType: "user", + }); + + await defaultEvents["approval.candidate"]!( + { + candidateId: "candidate-1", + outcome: "rejected", + requestId: "approval-1", + responderPrincipalId: "slack:T1:U777", + safeReason: "GitHub write access is required.", + sequence: 1, + stepIndex: 0, + turnId: "turn-1", + }, + channel, + ctx, + ); + + expect(postEphemeral).toHaveBeenCalledWith("U777", "GitHub write access is required."); + }); + + it("updates the shared card only after settlement", async () => { + const { channel, request } = buildChannelStub({ + pendingApprovalCards: { + "approval-1": { + actionId: "eve_input:approval-1:button:1", + messageBlocks: [ + { + actions: [{ action_id: "eve_input:approval-1:button:1" }], + body: { text: "Approve?", type: "mrkdwn" }, + type: "card", + }, + ], + messageTs: "123.456", + userId: "U777", + }, + }, + }); + + await defaultEvents["approval.settled"]!( + { + outcome: "approved", + requestId: "approval-1", + responderPrincipalId: "slack:T1:U777", + sequence: 1, + stepIndex: 0, + turnId: "turn-1", + }, + channel, + sessionCtx, + ); + + expect(request).toHaveBeenCalledWith( + "chat.update", + expect.objectContaining({ channel: "C123", text: "Answered: Approve", ts: "123.456" }), + ); + expect(channel.state.pendingApprovalCards).toEqual({}); + }); +}); + describe("defaultEvents authorization.required", () => { it("posts a public status and delivers the challenge ephemerally to the triggering user", async () => { const { channel, post, postEphemeral } = buildChannelStub({ triggeringUserId: "U777" }); @@ -68,6 +164,19 @@ describe("defaultEvents authorization.required", () => { expect(channel.state.pendingAuthMessageTs).toEqual({ notion: "ts1" }); }); + it("does not post a public status for candidate-scoped authorization", async () => { + const { channel, post, postEphemeral } = buildChannelStub({ triggeringUserId: "U777" }); + + await defaultEvents["authorization.required"]!( + { ...authRequiredEvent(), candidateId: "candidate-1" }, + channel, + sessionCtx, + ); + + expect(post).not.toHaveBeenCalled(); + expect(postEphemeral).toHaveBeenCalledTimes(1); + }); + it("targets the current Slack caller instead of stale channel state", async () => { const { channel, postEphemeral } = buildChannelStub({ triggeringUserId: "U_FIRST" }); const currentCaller = sessionContext({ diff --git a/packages/eve/src/public/channels/slack/defaults.ts b/packages/eve/src/public/channels/slack/defaults.ts index 48008eaa2..9ae7c50bc 100644 --- a/packages/eve/src/public/channels/slack/defaults.ts +++ b/packages/eve/src/public/channels/slack/defaults.ts @@ -11,6 +11,7 @@ import { type ConnectionAuthorizationOutcome, } from "#public/channels/slack/connections.js"; import { + buildAnsweredBlocks, formatInputRequestFallbackText, renderInputRequestBlocks, } from "#public/channels/slack/hitl.js"; @@ -32,6 +33,21 @@ const log = createLogger("slack.defaults"); const REASONING_TYPING_REFRESH_INTERVAL_MS = 5_000; const REASONING_TYPING_MIN_PROGRESS_CHARS = 4; +function blockContainsAction(block: unknown, actionId: string): boolean { + if (typeof block !== "object" || block === null) return false; + const candidate = block as { actions?: unknown; elements?: unknown }; + return [candidate.actions, candidate.elements].some( + (entries) => + Array.isArray(entries) && + entries.some( + (entry) => + typeof entry === "object" && + entry !== null && + (entry as { action_id?: unknown }).action_id === actionId, + ), + ); +} + /** * Workspace-scoped projection of the Slack actor that produced * `message`, derived into a {@link SessionAuthContext}. Used by both @@ -146,6 +162,67 @@ function buildInputRequestPosts( * which user overrides cannot express. */ export const defaultEvents: SlackChannelInternalEvents = { + async "approval.candidate"(event, channel, ctx) { + const currentUserId = slackUserIdFromAuthContext(ctx.session.auth.current); + if (event.outcome === "pending" && currentUserId !== undefined) { + channel.state.pendingApprovalCandidateUsers = { + ...channel.state.pendingApprovalCandidateUsers, + [event.candidateId]: currentUserId, + }; + await channel.thread.postEphemeral( + currentUserId, + "Checking whether you can approve this action…", + ); + return; + } + const mappedUserId = channel.state.pendingApprovalCandidateUsers?.[event.candidateId]; + const userId = + mappedUserId ?? + (ctx.session.auth.current?.principalId === event.responderPrincipalId + ? currentUserId + : undefined); + if (userId === undefined) return; + if (event.outcome === "rejected" || event.outcome === "failed") { + await channel.thread.postEphemeral( + userId, + event.safeReason ?? "We couldn’t verify your approval. Please try again.", + ); + const next = { ...channel.state.pendingApprovalCandidateUsers }; + delete next[event.candidateId]; + channel.state.pendingApprovalCandidateUsers = next; + } + }, + + async "approval.settled"(event, channel, _ctx) { + const cards = channel.state.pendingApprovalCards ?? {}; + const card = cards[event.requestId]; + if (card === undefined || channel.state.channelId === null) return; + const answerLabel = event.outcome === "approved" ? "Approve" : "Cancel"; + const blocks = card.messageBlocks.flatMap((block) => { + if (!blockContainsAction(block, card.actionId)) return [block]; + if (typeof block !== "object" || block === null) return []; + const candidate = block as Record; + if (candidate.type !== "card") { + return buildAnsweredBlocks({ answerLabel, promptBlocks: [], userId: card.userId }); + } + const { actions: _actions, ...withoutActions } = candidate; + return buildAnsweredBlocks({ + answerLabel, + promptBlocks: [withoutActions], + userId: card.userId, + }); + }); + await channel.slack.request("chat.update", { + blocks, + channel: channel.state.channelId, + text: `Answered: ${answerLabel}`, + ts: card.messageTs, + }); + const next = { ...cards }; + delete next[event.requestId]; + channel.state.pendingApprovalCards = next; + }, + async "turn.started"(_event, channel, _ctx) { channel.state.pendingToolCallMessage = null; channel.state.lastReasoningTypingAtMs = null; @@ -239,7 +316,7 @@ export const defaultEvents: SlackChannelInternalEvents = { // the session is blocked and later see it complete. The challenge // itself remains private. const pending = channel.state.pendingAuthMessageTs ?? {}; - if (pending[event.name] === undefined) { + if (event.candidateId === undefined && pending[event.name] === undefined) { const publicText = buildAuthRequiredPublicText({ displayName, hasUser: triggeringUserId !== null, @@ -289,7 +366,7 @@ export const defaultEvents: SlackChannelInternalEvents = { async "authorization.completed"(event, channel, _ctx) { const displayName = event.authorization?.displayName ?? formatConnectionDisplayName(event.name); - if (event.outcome === "authorized") { + if (event.outcome === "authorized" && event.candidateId === undefined) { await channel.thread.startTyping(`Connected to ${displayName}. Resuming...`); } diff --git a/packages/eve/src/public/channels/slack/interactions.ts b/packages/eve/src/public/channels/slack/interactions.ts index b8ca0d119..0d575712c 100644 --- a/packages/eve/src/public/channels/slack/interactions.ts +++ b/packages/eve/src/public/channels/slack/interactions.ts @@ -42,10 +42,6 @@ import { isHitlAction, type HitlFreeformModalMetadata, } from "#public/channels/slack/hitl.js"; -import { - SLACK_CARD_SUBTEXT_MAX_LENGTH, - truncateCardSubtext, -} from "#public/channels/slack/limits.js"; import type { SlackChannelConfig, SlackChannelState, @@ -180,6 +176,23 @@ function extractActionLabel(action: Record): string | undefined return undefined; } +function buildPendingApprovalCards( + interaction: ParsedBlockActionsPayload, +): NonNullable { + const cards: NonNullable = {}; + for (const action of interaction.actions) { + const response = deriveHitlResponse(action); + if (response === null || !action.messageTs) continue; + cards[response.requestId] = { + actionId: action.actionId, + messageBlocks: interaction.messageBlocks, + messageTs: action.messageTs, + userId: action.user.id, + }; + } + return cards; +} + function findPromptBlock(blocks: readonly unknown[]): unknown { return findPromptBlocks(blocks)[0]; } @@ -207,117 +220,6 @@ function readPromptTextFromBlocks(blocks: readonly unknown[]): string | undefine return typeof text === "string" && text.length > 0 ? text : undefined; } -function buildAnsweredHitlMessageBlocks(input: { - readonly actionId: string; - readonly answerLabel: string; - readonly messageBlocks: readonly unknown[]; - readonly userId: string; -}): unknown[] { - const actionBlockIndex = findActionBlockIndex(input.messageBlocks, input.actionId); - if (actionBlockIndex === -1) { - return buildAnsweredBlocks({ - promptBlocks: findPromptBlocks(input.messageBlocks), - answerLabel: input.answerLabel, - userId: input.userId, - }); - } - - const actionBlock = input.messageBlocks[actionBlockIndex]; - const answeredBlocks = - answeredBlocksFromActionBlock({ - answerLabel: input.answerLabel, - block: actionBlock, - userId: input.userId, - }) ?? - buildAnsweredBlocks({ - promptBlocks: promptBlocksFromActionBlock(actionBlock), - answerLabel: input.answerLabel, - userId: input.userId, - }); - return [ - ...input.messageBlocks.slice(0, actionBlockIndex), - ...answeredBlocks, - ...input.messageBlocks.slice(actionBlockIndex + 1), - ]; -} - -function findActionBlockIndex(blocks: readonly unknown[], actionId: string): number { - return blocks.findIndex((block) => blockContainsActionId(block, actionId)); -} - -function blockContainsActionId(block: unknown, actionId: string): boolean { - if (!isObjectRecord(block)) return false; - return ( - actionsContainActionId(block.elements, actionId) || - actionsContainActionId(block.actions, actionId) - ); -} - -function actionsContainActionId(actions: unknown, actionId: string): boolean { - if (!Array.isArray(actions)) return false; - return actions.some((element) => isObjectRecord(element) && element.action_id === actionId); -} - -function isObjectRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null; -} - -function answeredBlocksFromActionBlock(input: { - readonly answerLabel: string; - readonly block: unknown; - readonly userId: string; -}): unknown[] | undefined { - if (!isObjectRecord(input.block) || input.block.type !== "card") return undefined; - - const { actions: _actions, subtext: _subtext, ...blockWithoutActions } = input.block; - const answeredCard = { - ...blockWithoutActions, - subtext: { - type: "mrkdwn", - text: formatAnsweredCardSubtext(input), - verbatim: false, - }, - }; - return hasCardContent(answeredCard) ? [answeredCard] : undefined; -} - -const ANSWERED_CARD_SUBTEXT_PREFIX = ":white_check_mark: *"; -const ANSWERED_CARD_SUBTEXT_SUFFIX = "*"; - -function formatAnsweredCardSubtext(input: { - readonly answerLabel: string; - readonly userId: string; -}): string { - const attribution = input.userId.length > 0 ? ` by <@${input.userId}>` : ""; - const labelBudget = - SLACK_CARD_SUBTEXT_MAX_LENGTH - - ANSWERED_CARD_SUBTEXT_PREFIX.length - - ANSWERED_CARD_SUBTEXT_SUFFIX.length - - attribution.length; - const label = truncateWithEllipsis(input.answerLabel, labelBudget); - return truncateCardSubtext( - `${ANSWERED_CARD_SUBTEXT_PREFIX}${label}${ANSWERED_CARD_SUBTEXT_SUFFIX}${attribution}`, - ); -} - -function truncateWithEllipsis(value: string, maxLength: number): string { - if (maxLength <= 0) return ""; - if (value.length <= maxLength) return value; - const sliceLength = Math.max(0, maxLength - 3); - return `${value.slice(0, sliceLength).trimEnd()}...`; -} - -function promptBlocksFromActionBlock(block: unknown): unknown[] { - if (!isObjectRecord(block) || block.type !== "card") return []; - - const { actions: _actions, ...blockWithoutActions } = block; - return hasCardContent(blockWithoutActions) ? [blockWithoutActions] : []; -} - -function hasCardContent(block: Record): boolean { - return block.body !== undefined || block.title !== undefined || block.hero_image !== undefined; -} - /** * Channel-supplied dependencies for {@link handleInteractionPost}. * @@ -385,7 +287,7 @@ export async function handleInteractionPost( ctx.waitUntil( ctx .send( - { inputResponses }, + { inputResponses, pendingApprovalCards: buildPendingApprovalCards(interaction) }, { auth: buildSlackAuthContext({ channelId: interaction.channelId, @@ -400,6 +302,7 @@ export async function handleInteractionPost( threadTs: interaction.threadTs, teamId: interaction.teamId ?? null, triggeringUserId: user.id, + pendingApprovalCards: buildPendingApprovalCards(interaction), }, }, ) @@ -407,12 +310,6 @@ export async function handleInteractionPost( log.error("HITL interaction delivery failed", { error }); }), ); - - ctx.waitUntil( - updateAnsweredHitlCard(interaction, deps).catch((error: unknown) => { - log.error("HITL answered-card update failed", { error }); - }), - ); } const onInteraction = deps.config.onInteraction; @@ -589,42 +486,6 @@ async function handleViewSubmission( return ack; } -async function updateAnsweredHitlCard( - interaction: ParsedBlockActionsPayload, - deps: InteractionHandlerDeps, -): Promise { - const hitlAction = interaction.actions.find((a) => isHitlAction(a.actionId)); - if (!hitlAction || !hitlAction.messageTs) return; - - const answerLabel = hitlAction.label ?? hitlAction.selectedOptionValue ?? hitlAction.value; - if (!answerLabel) return; - - const blocks = buildAnsweredHitlMessageBlocks({ - actionId: hitlAction.actionId, - answerLabel, - messageBlocks: interaction.messageBlocks, - userId: hitlAction.user.id, - }); - - const token = await resolveSlackBotToken(deps.config.credentials?.botToken); - const response = await fetch("https://slack.com/api/chat.update", { - method: "POST", - headers: { - authorization: `Bearer ${token}`, - "content-type": "application/json; charset=utf-8", - }, - body: JSON.stringify({ - channel: interaction.channelId, - ts: hitlAction.messageTs, - blocks, - text: `Answered: ${answerLabel}`, - }), - }); - if (!response.ok) { - throw new Error(`Slack chat.update returned HTTP ${response.status}`); - } -} - async function updateAnsweredFreeformCard(input: { readonly channelId: string; readonly messageTs: string; diff --git a/packages/eve/src/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index c96b5baa7..13f920a34 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.test.ts @@ -8,7 +8,7 @@ import { isCompiledChannel, type CompiledChannel } from "#channel/compiled-chann import { isHttpRouteDefinition } from "#channel/routes.js"; import { ContextContainer, contextStorage } from "#context/container.js"; import { SessionKey } from "#context/keys.js"; -import type { UnstampedMessageStreamEvent } from "#protocol/message.js"; +import type { HandleMessageStreamEvent } from "#protocol/message.js"; import { decodeSlackApiBody } from "#public/channels/slack/api-encoding.js"; import { HITL_ACTION_PREFIX, @@ -18,7 +18,6 @@ import { } from "#public/channels/slack/hitl.js"; import { SLACK_CARD_BODY_TEXT_MAX_LENGTH, - SLACK_CARD_SUBTEXT_MAX_LENGTH, SLACK_MAX_BLOCKS_PER_MESSAGE, SLACK_MESSAGE_TEXT_MAX_LENGTH, SLACK_SECTION_TEXT_MAX_LENGTH, @@ -78,9 +77,9 @@ const stubAlsContext = (() => { function callEvent( adapter: ChannelAdapter, - event: UnstampedMessageStreamEvent, + event: HandleMessageStreamEvent, ctx: any, -): Promise { +): Promise { return contextStorage.run(stubAlsContext, () => callAdapterEventHandler(adapter, event, ctx)); } @@ -115,11 +114,11 @@ function captureAccessor(initialContinuationToken: string): { }; } -function makeEvent( +function makeEvent( type: T, data: unknown, -): UnstampedMessageStreamEvent { - return { type, data } as UnstampedMessageStreamEvent; +): HandleMessageStreamEvent { + return { type, data } as HandleMessageStreamEvent; } const THREAD_STATE = { @@ -2177,8 +2176,9 @@ describe("slackChannel() HITL interaction pipeline", () => { expect(send).toHaveBeenCalledTimes(1); const [payload, options] = send.mock.calls[0]!; - expect(payload).toEqual({ + expect(payload).toMatchObject({ inputResponses: [{ optionId: "approve", requestId: "approval_abc123" }], + pendingApprovalCards: { approval_abc123: expect.objectContaining({ userId: "U_APPROVER" }) }, }); expect(options).toMatchObject({ auth: { @@ -2304,47 +2304,14 @@ describe("slackChannel() HITL interaction pipeline", () => { ); expect(send).toHaveBeenCalledTimes(1); - expect(send.mock.calls[0]?.[0]).toEqual({ + expect(send.mock.calls[0]?.[0]).toMatchObject({ inputResponses: [{ optionId: "approve", requestId: "approval_451" }], + pendingApprovalCards: { approval_451: expect.any(Object) }, }); - const updateCall = fetchMock.mock.calls.find( - ([url]) => String(url) === "https://slack.com/api/chat.update", - ); - expect(updateCall).toBeDefined(); - const body = parseSlackRequestBody(updateCall?.[1] as RequestInit) as { - blocks: Array<{ - actions?: Array<{ action_id?: string }>; - elements?: Array<{ action_id?: string }>; - subtext?: { text?: string }; - text?: { text?: string }; - }>; - channel: string; - text: string; - ts: string; - }; - - expect(body).toMatchObject({ - channel: "C01", - text: "Answered: Approve", - ts: "1700000000.000010", - }); - expect(JSON.stringify(body.blocks)).toContain("Approve issue 451?"); - expect(JSON.stringify(body.blocks)).toContain("Approve"); - expect(JSON.stringify(body.blocks)).toContain("Tool input"); - expect(JSON.stringify(body.blocks)).toContain("Approve issue 508?"); - expect(body.blocks[0]?.subtext?.text).toBe(":white_check_mark: *Approve* by <@U_APPROVER>"); - expect(body.blocks[0]?.subtext?.text?.length).toBeLessThanOrEqual( - SLACK_CARD_SUBTEXT_MAX_LENGTH, - ); - - const remainingActionIds = body.blocks.flatMap( - (block) => - (block.actions ?? block.elements) - ?.map((element) => element.action_id) - .filter((actionId): actionId is string => typeof actionId === "string") ?? [], - ); - expect(remainingActionIds).toEqual([secondDenyActionId, secondApproveActionId]); + expect( + fetchMock.mock.calls.find(([url]) => String(url) === "https://slack.com/api/chat.update"), + ).toBeUndefined(); }); it("covers the observed e0 batched escalation approval run", async () => { @@ -2448,36 +2415,14 @@ describe("slackChannel() HITL interaction pipeline", () => { ); expect(send).toHaveBeenCalledTimes(1); - expect(send.mock.calls[0]?.[0]).toEqual({ + expect(send.mock.calls[0]?.[0]).toMatchObject({ inputResponses: [{ optionId: "cancel", requestId: "approval_451" }], + pendingApprovalCards: { approval_451: expect.any(Object) }, }); - const updateCall = fetchMock.mock.calls.find( - ([url]) => String(url) === "https://slack.com/api/chat.update", - ); - expect(updateCall).toBeDefined(); - const body = parseSlackRequestBody(updateCall?.[1] as RequestInit) as { - blocks: Array<{ - actions?: Array<{ action_id?: string }>; - child_blocks?: Array<{ text?: { text?: string } }>; - elements?: Array<{ action_id?: string }>; - subtext?: { text?: string }; - }>; - }; - - expect(body.blocks[0]?.subtext?.text).toBe(":white_check_mark: *Cancel* by <@U0AT7H56S90>"); - expect(body.blocks[1]?.child_blocks?.[0]?.text?.text).toContain('"issueNumber": 451'); - expect(body.blocks[3]?.child_blocks?.[0]?.text?.text).toContain('"issueNumber": 508'); - const remainingActionIds = body.blocks.flatMap( - (block) => - (block.actions ?? block.elements) - ?.map((element) => element.action_id) - .filter((actionId): actionId is string => typeof actionId === "string") ?? [], - ); - expect(remainingActionIds).toEqual([ - `${HITL_ACTION_PREFIX}approval_508:button:0`, - `${HITL_ACTION_PREFIX}approval_508:button:1`, - ]); + expect( + fetchMock.mock.calls.find(([url]) => String(url) === "https://slack.com/api/chat.update"), + ).toBeUndefined(); }); it("resumes freeform modal answers with the submitting Slack user auth", async () => { diff --git a/packages/eve/src/public/channels/slack/slackChannel.ts b/packages/eve/src/public/channels/slack/slackChannel.ts index e64566e71..aa6a8e932 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -1,5 +1,6 @@ import { parseSlackWebhookBody } from "#compiled/@chat-adapter/slack/webhook.js"; +import { defaultDeliverResult } from "#channel/adapter.js"; import type { CrossChannelReceiveOptions } from "#channel/cross-channel-receive.js"; import type { Session, SessionHandle } from "#channel/session.js"; import type { CancelTurnResult, SessionAuthContext } from "#channel/types.js"; @@ -8,7 +9,7 @@ import type { SessionContext } from "#public/definitions/callback-context.js"; import type { ChannelSessionOps } from "#public/definitions/channel.js"; import { createLogger, logError } from "#internal/logging.js"; -import type { UnstampedMessageStreamEvent } from "#protocol/message.js"; +import type { HandleMessageStreamEvent } from "#protocol/message.js"; import { buildSlackBinding, buildSlackWorkspaceHandle, @@ -69,8 +70,8 @@ import { markEventHandled } from "./utils.js"; const log = createLogger("slack.channel"); -type EventData = - Extract extends { data: infer D } ? D : undefined; +type EventData = + Extract extends { data: infer D } ? D : undefined; /** * Base Slack context for inbound webhook handlers. These hooks run before the @@ -111,7 +112,7 @@ export type { } from "#public/channels/slack/api.js"; export type { SlackWebhookVerifier } from "#public/channels/slack/verify.js"; -type SlackEventHandler = ( +type SlackEventHandler = ( data: EventData, channel: SlackEventContext, ctx: SessionContext, @@ -167,6 +168,13 @@ type SlackSessionFailedHandler = ( * step boundaries. Anything written here must round-trip through * `JSON.stringify` / `JSON.parse`. */ +export interface SlackPendingApprovalCard { + readonly actionId: string; + readonly messageBlocks: readonly unknown[]; + readonly messageTs: string; + readonly userId: string; +} + export interface SlackChannelState { /** Slack channel id seeded by the inbound mention. */ channelId: string | null; @@ -204,6 +212,8 @@ export interface SlackChannelState { * resolution outcome. */ pendingAuthMessageTs?: Record; + pendingApprovalCards?: Record; + pendingApprovalCandidateUsers?: Record; } /** @@ -448,6 +458,8 @@ export type SlackInboundResultOrPromise = SlackMentionResultOrPromise; * {@link SessionContext}; `session.failed` receives only data and context. */ export interface SlackChannelEvents { + readonly "approval.candidate"?: SlackEventHandler<"approval.candidate">; + readonly "approval.settled"?: SlackEventHandler<"approval.settled">; readonly "turn.started"?: SlackEventHandler<"turn.started">; readonly "actions.requested"?: SlackEventHandler<"actions.requested">; readonly "action.result"?: SlackEventHandler<"action.result">; @@ -686,6 +698,8 @@ export function slackChannel(config: SlackChannelConfig = {}): SlackChannel { lastReasoningTypingAtMs: null, lastReasoningTypingStatus: null, pendingAuthMessageTs: {}, + pendingApprovalCards: {}, + pendingApprovalCandidateUsers: {}, }, fetchFile: slackFetchFile, metadata(state): SlackInstrumentationMetadata { @@ -701,6 +715,17 @@ export function slackChannel(config: SlackChannelConfig = {}): SlackChannel { return rebuildSlackContext(state, session, config.credentials); }, + deliver(payload, channel) { + const incoming = payload["pendingApprovalCards"]; + if (typeof incoming === "object" && incoming !== null) { + channel.state.pendingApprovalCards = { + ...channel.state.pendingApprovalCards, + ...(incoming as SlackChannelState["pendingApprovalCards"]), + }; + } + return defaultDeliverResult(payload); + }, + routes: [ POST( config.route ?? SLACK_CHANNEL_DEFAULT_ROUTE, diff --git a/packages/eve/src/public/definitions/channel.ts b/packages/eve/src/public/definitions/channel.ts index 7e0b09579..07977c065 100644 --- a/packages/eve/src/public/definitions/channel.ts +++ b/packages/eve/src/public/definitions/channel.ts @@ -16,7 +16,7 @@ import type { RunInput, } from "#channel/types.js"; import { buildCallbackContext } from "#context/build-callback-context.js"; -import type { UnstampedMessageStreamEvent, MessageStreamEvent } from "#protocol/message.js"; +import type { HandleMessageStreamEvent } from "#protocol/message.js"; import type { SessionContext } from "#public/definitions/callback-context.js"; import type { GenericChannelDefinition, GenericReceiveInput } from "#shared/channel-definition.js"; @@ -160,7 +160,7 @@ export interface Agent { getEventStream( sessionId: string, options?: GetEventStreamOptions, - ): Promise>; + ): Promise>; } /** @@ -201,8 +201,8 @@ export function isDisabledRouteSentinel(value: unknown): value is DisabledRouteS ); } -type EventData = - Extract extends { data: infer D } ? D : undefined; +type EventData = + Extract extends { data: infer D } ? D : undefined; /** * Session operations on the `channel` argument of every channel event handler. @@ -218,7 +218,7 @@ export interface ChannelSessionOps { */ export type ChannelContext = TCtx & ChannelSessionOps; -type ChannelEventHandler = ( +type ChannelEventHandler = ( data: EventData, channel: ChannelContext, ctx: SessionContext, @@ -236,6 +236,8 @@ type ChannelSessionFailedHandler = ( * and the channel context, with no `ctx`. */ export interface ChannelEvents { + readonly "approval.candidate"?: ChannelEventHandler<"approval.candidate", TCtx>; + readonly "approval.settled"?: ChannelEventHandler<"approval.settled", TCtx>; readonly "turn.started"?: ChannelEventHandler<"turn.started", TCtx>; readonly "actions.requested"?: ChannelEventHandler<"actions.requested", TCtx>; readonly "action.result"?: ChannelEventHandler<"action.result", TCtx>; @@ -339,6 +341,8 @@ export function defineChannel< // The Record type fails to compile if this map drifts from the ChannelEvents // keys in either direction. const channelEventTypes: Record = { + "approval.candidate": null, + "approval.settled": null, "turn.started": null, "actions.requested": null, "action.result": null, @@ -429,8 +433,8 @@ function buildAdapter = Extract< - MessageStreamEvent, +type ProtocolEvent = Extract< + HandleMessageStreamEvent, { type: TType } >; @@ -17,6 +16,8 @@ type ProtocolEvent = Extract< */ export interface HookEventMap { readonly "action.result": ProtocolEvent<"action.result">; + readonly "approval.candidate": ProtocolEvent<"approval.candidate">; + readonly "approval.settled": ProtocolEvent<"approval.settled">; readonly "actions.requested": ProtocolEvent<"actions.requested">; readonly "authorization.completed": ProtocolEvent<"authorization.completed">; readonly "authorization.required": ProtocolEvent<"authorization.required">; diff --git a/packages/eve/src/shared/channel-definition.ts b/packages/eve/src/shared/channel-definition.ts index 043c9970c..42543a5b0 100644 --- a/packages/eve/src/shared/channel-definition.ts +++ b/packages/eve/src/shared/channel-definition.ts @@ -1,7 +1,8 @@ import { type ChannelCors } from "#channel/cors.js"; import type { RouteDefinition, SendFn } from "#channel/routes.js"; import type { Session, SessionHandle } from "#channel/session.js"; -import type { SessionAuthContext } from "#channel/types.js"; +import type { DeliverPayload, SessionAuthContext } from "#channel/types.js"; +import type { StepInput } from "#harness/types.js"; /** * Enriched return shape from a channel's {@link ChannelAdapter.fetchFile} @@ -70,6 +71,7 @@ export interface GenericChannelDefinition< context?(state: NonNullable, session: SessionHandle): TCtx; readonly routes: readonly RouteDefinition[]; + deliver?(payload: DeliverPayload, ctx: TCtx): StepInput | void | Promise; receive?( input: GenericReceiveInput, args: { send: SendFn },