diff --git a/apps/ai/src/chat/ChatSession.test.ts b/apps/ai/src/chat/ChatSession.test.ts index 96cbdfd2cc..7b391e02f2 100644 --- a/apps/ai/src/chat/ChatSession.test.ts +++ b/apps/ai/src/chat/ChatSession.test.ts @@ -357,7 +357,7 @@ describe("ChatSession turn mutex", () => { const second = turnId()! // The aborted turn keeps draining and eventually calls `endTurn` for its own message. - session.endTurn(first) + session.endTurn(first, { reason: "stop" }) assert.notEqual(first, second) assert.isTrue(session.running(), "the second turn still owns the slot") @@ -402,6 +402,82 @@ describe("ChatSession turn mutex", () => { }) }) +describe("ChatSession turn slot transitions", () => { + const begin = (session: ChatSession, messageId = "m1") => + session.beginTurn({ sessionId: "org_test:tab", messageId, text: "hi", tenant: TENANT, origin: APP }) + const turnEnds = (session: ChatSession, messageId: string | undefined) => + Arr.filter(session.since(0), (event) => event.type === "turn-end" && event.messageId === messageId) + + it("reads a stale claim without expiring it; only a transition does", () => { + const { session, state, turnId } = makeSession() + begin(session) + const turn = turnId() + state.storage.sql.exec( + "UPDATE session SET running_since = ? WHERE id = 1", + Date.now() - TURN_STALE_MS - 1, + ) + + assert.isFalse(session.running()) + assert.isFalse(session.holdsTurn(turn!)) + assert.lengthOf(turnEnds(session, turn), 0, "a read appends nothing") + + session.alarm() + assert.lengthOf(turnEnds(session, turn), 1) + assert.strictEqual(turnId(), undefined) + }) + + it("treats a legacy claim with no start time as stale, so a new turn is admitted", () => { + const { session, state } = makeSession() + state.storage.sql.exec( + "UPDATE session SET running = 1, running_since = NULL, running_message_id = NULL WHERE id = 1", + ) + + assert.isFalse(session.running()) + assert.isDefined(begin(session), "the legacy claim no longer wedges the conversation") + }) + + it("stops the heartbeat on a legacy claim instead of re-arming it forever", () => { + const { session, state } = makeSession() + state.storage.sql.exec( + "UPDATE session SET running = 1, running_since = NULL, running_message_id = 'legacy' WHERE id = 1", + ) + + session.alarm() + + assert.lengthOf(state.alarms, 0) + assert.lengthOf(turnEnds(session, "legacy"), 1) + assert.isFalse(session.running()) + }) + + it("writes one turn-end when the turn ends and is aborted at once", () => { + const { session, turnId } = makeSession() + begin(session) + const turn = turnId()! + + session.abort("user") + const ended = session.endTurn(turn, { reason: "stop" }) + + assert.isFalse(ended) + const ends = turnEnds(session, turn) + assert.lengthOf(ends, 1) + assert.strictEqual(ends[0]?.type === "turn-end" ? ends[0].reason : undefined, "aborted") + }) + + it("writes one turn-end when an ended turn is then aborted or ended again", () => { + const { session, turnId } = makeSession() + begin(session) + const turn = turnId()! + + assert.isTrue(session.endTurn(turn, { reason: "max-steps" })) + session.abort("user") + assert.isFalse(session.endTurn(turn, { reason: "error", error: "late" })) + + const ends = turnEnds(session, turn) + assert.lengthOf(ends, 1) + assert.strictEqual(ends[0]?.type === "turn-end" ? ends[0].reason : undefined, "max-steps") + }) +}) + describe("ChatSession turn heartbeat", () => { /** * An outbound fetch never keeps a Durable Object alive, so an autonomous turn nobody was watching @@ -432,7 +508,7 @@ describe("ChatSession turn heartbeat", () => { tenant: TENANT, origin: APP, }) - session.endTurn(turnId()!) + session.endTurn(turnId()!, { reason: "stop" }) session.alarm() assert.lengthOf(state.alarms, 1) @@ -626,6 +702,7 @@ describe("ChatSession abort interrupts the running turn", () => { ) assert.isFalse(session.running()) + session.alarm() assert.strictEqual((await span)?.attributes.get("maple.chat.abort_reason"), "stale") }) diff --git a/apps/ai/src/chat/ChatSession.ts b/apps/ai/src/chat/ChatSession.ts index 75ec0856b5..2cf1da5bdb 100644 --- a/apps/ai/src/chat/ChatSession.ts +++ b/apps/ai/src/chat/ChatSession.ts @@ -176,6 +176,34 @@ export const MAX_TURN_RESUMES = 2 /** What the dead attempt's assistant message is closed with when the turn is started again. */ const TURN_RESTARTED = "Maple was restarted mid-review and is starting this review again." +/** How a conversation turn ended: its one `turn-end`, without the message it lands on. */ +export type TurnEnding = Omit< + Extract, + "type" | "messageId" | "task" +> + +const TURN_FAILED: TurnEnding = { reason: "error", error: CHAT_TURN_FAILED } +const TURN_ABANDONED: TurnEnding = { + reason: "error", + error: "The turn stopped responding and was abandoned.", +} + +/** The turn slot. `Stale` is a claim past {@link TURN_STALE_MS}, or a legacy one with no start time or id. */ +type TurnSlot = + | { readonly _tag: "Idle" } + | { readonly _tag: "Held"; readonly messageId: string } + | { readonly _tag: "Stale"; readonly messageId: string | undefined } + +const IDLE: TurnSlot = { _tag: "Idle" } + +/** The changes {@link ChatSession.transition} makes to the slot, each compared against it first. */ +type SlotChange = + | { readonly _tag: "Begin"; readonly messageId: string; readonly input: TurnInput } + | { readonly _tag: "End"; readonly messageId: string; readonly ending: TurnEnding } + | { readonly _tag: "Abort"; readonly reason: ChatAbortReason } + | { readonly _tag: "Expire" } + | { readonly _tag: "Resume"; readonly from: string; readonly to: string; readonly resumes: number } + /** * What an evicted turn needs to run again: everything `beginTurn` was given. Plain JSON in the * session row, decoded back through the same schemas that describe it on the wire. @@ -307,50 +335,85 @@ export class ChatSession { } /** - * The message id of the turn that holds the slot *right now*, or `undefined` if none does. - * - * Expiring a stale claim here rather than in a separate sweep means every caller — `beginTurn`, - * the turn's own between-step check — agrees on the answer without needing an alarm to have - * fired first. - * - * One read serves both questions callers ask ("is anything running?" and "is *this* running?"). - * They used to be separate methods issuing the same SELECT, and `holdsTurn` — called once per - * streamed event — paid for both. + * The turn slot, read and nothing else. Called once per streamed event, so it never writes: + * expiring a stale claim is a {@link transition}, taken by `beginTurn` and the alarm. */ - private runningTurn(): string | undefined | null { + private slot(): TurnSlot { const row = this.sessionRow() - if (row.running !== 1) return undefined - if (row.running_since !== null && Date.now() - row.running_since > TURN_STALE_MS) { - const messageId = row.running_message_id - this.clearRunning() - if (messageId !== null) { + if (row.running !== 1) return IDLE + // A row with no start time or no id (pre-watchdog) can never be judged fresh, so it is stale. + if ( + row.running_since === null || + row.running_message_id === null || + Date.now() - row.running_since > TURN_STALE_MS + ) { + return { _tag: "Stale", messageId: row.running_message_id ?? undefined } + } + return { _tag: "Held", messageId: row.running_message_id } + } + + /** + * Every change to the turn slot, each a compare-and-set, and the only writer of a turn's + * `turn-end`: leaving the slot and appending it are one synchronous step, so a turn gets one. + */ + private transition(change: SlotChange): boolean { + const slot = this.slot() + switch (change._tag) { + case "Begin": { + if (slot._tag !== "Idle") return false + this.sql.exec( + "UPDATE session SET running = 1, running_since = ?, running_message_id = ?, running_input = ?, running_resumes = 0 WHERE id = 1", + Date.now(), + change.messageId, + encodeTurnInput(change.input), + ) + return true + } + case "End": { + if (slot._tag === "Idle" || slot.messageId !== change.messageId) return false + this.release(slot.messageId, change.ending) + return true + } + case "Abort": { + if (slot._tag === "Idle") return false + this.release(slot.messageId, { reason: "aborted" }, change.reason) + return true + } + case "Expire": { + if (slot._tag !== "Stale") return false + this.release(slot.messageId, TURN_ABANDONED, "stale") + return true + } + case "Resume": { + if (slot._tag !== "Held" || slot.messageId !== change.from) return false this.append({ type: "turn-end", - messageId, + messageId: change.from, reason: "error", - error: "The turn stopped responding and was abandoned.", + error: TURN_RESTARTED, }) - this.interruptTurn(messageId, "stale") + this.sql.exec( + "UPDATE session SET running_since = ?, running_message_id = ?, running_resumes = ? WHERE id = 1", + Date.now(), + change.to, + change.resumes, + ) + return true } - return undefined } - // `null` means running but with no recorded id — pre-watchdog rows only. - return row.running_message_id } - private isRunning(): boolean { - return this.runningTurn() !== undefined - } - - private clearRunning(): void { + /** + * Clear the slot and close the turn that held it. A legacy claim with no id names no message a + * client is rendering, so there is no `turn-end` to write for it. + */ + private release(messageId: string | undefined, ending: TurnEnding, interrupt?: ChatAbortReason): void { this.sql.exec( "UPDATE session SET running = 0, running_since = NULL, running_message_id = NULL, running_input = NULL, running_resumes = NULL WHERE id = 1", ) - } - - /** The message id of the turn currently holding the slot, if any. */ - private runningMessageId(): string | undefined { - return this.runningTurn() ?? undefined + if (messageId === undefined) return + this.append({ type: "turn-end", messageId, ...ending }) + if (interrupt !== undefined) this.interruptTurn(messageId, interrupt) } /** @@ -483,7 +546,7 @@ export class ChatSession { /** Who is driving the turn, stated by whoever raised it. */ readonly origin: ChatTurnOrigin }): { cursor: number; messageId: string; turnMessageId: string } | undefined { - if (this.isRunning()) return undefined + this.transition({ _tag: "Expire" }) const cursor = this.cursor() // The assistant's message needs an id of its OWN. Reusing the user's meant `history()` found // the user message when opening the assistant one, so every text delta was appended to the @@ -497,12 +560,7 @@ export class ChatSession { tenant: input.tenant, origin: input.origin, } - this.sql.exec( - "UPDATE session SET running = 1, running_since = ?, running_message_id = ?, running_input = ?, running_resumes = 0 WHERE id = 1", - Date.now(), - turnId, - encodeTurnInput(turn), - ) + if (!this.transition({ _tag: "Begin", messageId: turnId, input: turn })) return undefined this.append({ type: "user-message", id: input.messageId, text: input.text }) this.startTurn(turnId, turn) return { cursor, messageId: input.messageId, turnMessageId: turnId } @@ -606,18 +664,19 @@ export class ChatSession { * effect promptly, and so a turn that has already been superseded stops writing. */ holdsTurn(messageId: string): boolean { - return this.runningTurn() === messageId + const slot = this.slot() + return slot._tag === "Held" && slot.messageId === messageId } /** - * Release the slot, but only if `messageId` still owns it. + * End `messageId`'s turn with `ending`, but only if it still owns the slot. Returns whether this + * call wrote the turn's `turn-end`; a later call for the same turn writes nothing. * * The compare-and-clear matters: an aborted turn keeps draining its in-flight model call for a * while, and a plain `running = 0` from that straggler would release a *newer* turn's claim. */ - endTurn(messageId: string): void { - if (this.runningMessageId() !== messageId) return - this.clearRunning() + endTurn(messageId: string, ending: TurnEnding): boolean { + return this.transition({ _tag: "End", messageId, ending }) } /** @@ -626,18 +685,11 @@ export class ChatSession { * Clears the claim, records the terminal event against the message that actually holds it, and * interrupts that turn's fiber, sub-agents included, rather than letting it run to its next event. * Taking the message id from the session rather than the caller is what makes the emitted - * `turn-end` land on the assistant message the client is rendering; a caller-supplied id named a - * message that never existed, so no client ever cleared its streaming state. + * `turn-end` land on the assistant message the client is rendering. */ abort(reason?: ChatAbortReason): void { - const messageId = this.runningTurn() - if (messageId === undefined) return - this.clearRunning() - if (messageId !== null) { - this.append({ type: "turn-end", messageId, reason: "aborted" }) - // Over RPC from another Worker, so not trusted to be one of ours. - this.interruptTurn(messageId, isChatAbortReason(reason) ? reason : "user") - } + // Over RPC from another Worker, so not trusted to be one of ours. + this.transition({ _tag: "Abort", reason: isChatAbortReason(reason) ? reason : "user" }) } /** Interrupt `messageId`'s fiber if this activation runs it; a straggler's id matches nothing. */ @@ -645,25 +697,30 @@ export class ChatSession { if (this.liveTurn?.messageId === messageId) this.liveTurn.controller.abort(reason) } + /** Whether a live claim holds the slot. A stale one does not: the next `beginTurn` expires it. */ running(): boolean { - return this.isRunning() + return this.slot()._tag === "Held" } /** * The heartbeat. Re-arms while this activation runs the turn that holds the slot. * - * A slot held by a turn this activation is not running means the object was evicted mid-turn (a - * deploy, or eviction before the heartbeat existed): the fiber is gone, so the slot is released - * with a terminal event now rather than when the 15-minute watchdog expires it. + * A stale claim is expired. A slot held by a turn this activation is not running means the + * object was evicted mid-turn (a deploy, or eviction before the heartbeat existed): the fiber is + * gone, so the slot is released with a terminal event now rather than when it goes stale. */ alarm(): void { - const messageId = this.runningTurn() - if (messageId === undefined) return - if (messageId !== null && this.liveTurn?.messageId !== messageId) { - this.recoverEvictedTurn(messageId) - return + const slot = this.slot() + switch (slot._tag) { + case "Idle": + return + case "Stale": + this.transition({ _tag: "Expire" }) + return + case "Held": + if (this.liveTurn?.messageId === slot.messageId) return this.armHeartbeat() + return this.recoverEvictedTurn(slot.messageId) } - this.armHeartbeat() } /** @@ -679,19 +736,13 @@ export class ChatSession { const turn = row.running_input === null ? Option.none() : decodeTurnInput(row.running_input) const resumes = row.running_resumes ?? 0 if (Option.isSome(turn) && resumable(turn.value) && resumes < MAX_TURN_RESUMES) { - this.append({ type: "turn-end", messageId, reason: "error", error: TURN_RESTARTED }) const turnId = SecureRandom.uuidV4Unsafe() - this.sql.exec( - "UPDATE session SET running_since = ?, running_message_id = ?, running_resumes = ? WHERE id = 1", - Date.now(), - turnId, - resumes + 1, - ) - this.startTurn(turnId, turn.value, { resume: resumes + 1 }) + if (this.transition({ _tag: "Resume", from: messageId, to: turnId, resumes: resumes + 1 })) { + this.startTurn(turnId, turn.value, { resume: resumes + 1 }) + } return } - this.clearRunning() - this.append({ type: "turn-end", messageId, reason: "error", error: CHAT_TURN_FAILED }) + if (!this.endTurn(messageId, TURN_FAILED)) return if (Option.isSome(turn) && resumable(turn.value)) { const input = { ...this.turnRunInput(messageId, turn.value), abandoned: true as const } this.ctx.waitUntil( @@ -758,33 +809,21 @@ export class ChatSession { * Drive one turn to completion, appending events as they are produced. * * Everything heavy — the Effect runtime, the service graph, the agent engine — is behind this - * dynamic import so none of it is evaluated at module scope. Failures are recorded as a - * terminal event rather than thrown: the log is what the client reads, so a turn that dies - * silently is indistinguishable from one that hung. + * dynamic import so none of it is evaluated at module scope. A runner that returns or fails + * without ending its turn is ended here as failed: the log is what the client reads, so a turn + * that dies silently is indistinguishable from one that hung. */ private runTurn(input: RunChatSessionTurnInput): Promise { const messageId = input.messageId return Effect.runPromise( - attempt("Failed to start turn runner", () => this.runner(input)).pipe( + attempt("Failed to run the turn", () => this.runner(input)).pipe( Effect.catchTag("@maple/ai/chat/ChatSessionError", (error) => - Effect.logError("[chat.turn] Failed to start turn runner", error.cause).pipe( - Effect.andThen( - Effect.sync(() => { - if (!this.holdsTurn(messageId)) return - this.append({ - type: "turn-end", - messageId, - reason: "error", - error: CHAT_TURN_FAILED, - }) - }), - ), - ), + Effect.logError("[chat.turn] Failed to run the turn", error.cause), ), Effect.ensuring( Effect.sync(() => { if (this.liveTurn?.messageId === messageId) this.liveTurn = undefined - this.endTurn(messageId) + this.endTurn(messageId, TURN_FAILED) }), ), ), @@ -828,7 +867,8 @@ export const chatSessionRpc = (session: ChatSession) => beginTurn: (input) => Effect.sync(() => session.beginTurn(input)), settleProposal: (input) => Effect.promise(() => session.settleProposal(input)), holdsTurn: (messageId) => Effect.sync(() => session.holdsTurn(messageId)), - endTurn: (messageId) => Effect.sync(() => session.endTurn(messageId)), + // A caller releasing a turn without saying how it ended is ending one that failed. + endTurn: (messageId) => Effect.sync(() => void session.endTurn(messageId, TURN_FAILED)), abort: (reason) => Effect.sync(() => session.abort(reason)), alarm: () => Effect.sync(() => session.alarm()), }) satisfies ChatSessionRpc diff --git a/apps/ai/src/chat/autonomous-outcome.test.ts b/apps/ai/src/chat/autonomous-outcome.test.ts new file mode 100644 index 0000000000..06084cbd93 --- /dev/null +++ b/apps/ai/src/chat/autonomous-outcome.test.ts @@ -0,0 +1,63 @@ +import { assert, describe, it } from "vitest" +import { makeAutonomousOutcome } from "./autonomous-outcome" + +describe("makeAutonomousOutcome", () => { + it("keeps an earlier phase's step limit in the final reason after a close-out reports", () => { + const outcome = makeAutonomousOutcome() + outcome.hold("max-steps") + outcome.record("pass", false) + outcome.hold("stop") + outcome.record("close-out", true) + + assert.deepEqual(outcome.ending("no report"), { reason: "max-steps" }) + assert.strictEqual(outcome.failure(), "step_limit") + }) + + it("ends on stop when no phase hit the step limit", () => { + const outcome = makeAutonomousOutcome() + outcome.hold("stop") + outcome.record("pass", true) + + assert.deepEqual(outcome.ending("no report"), { reason: "stop" }) + assert.strictEqual(outcome.failure(), undefined) + }) + + it("ends on the caller's error when no phase reported", () => { + const outcome = makeAutonomousOutcome() + outcome.record("pass", false, "model_error") + outcome.record("close-out", false, "time_limit") + + assert.deepEqual(outcome.ending("no report"), { reason: "error", error: "no report" }) + // The first reason is the one the pull request is told. + assert.strictEqual(outcome.failure(), "model_error") + }) + + it("discards the replaced attempt's failure once the pass restarts", () => { + const outcome = makeAutonomousOutcome() + outcome.record("pass", false, "stuck") + outcome.hold("max-steps") + outcome.record("restart", false) + + assert.strictEqual(outcome.failure(), "step_limit") + }) + + it("does not read a close-out's step limit as the pass's failure", () => { + const outcome = makeAutonomousOutcome() + outcome.hold("stop") + outcome.record("pass", false) + outcome.fail("answered_in_prose") + outcome.hold("max-steps") + outcome.record("close-out", false) + + assert.strictEqual(outcome.failure(), "answered_in_prose") + }) + + it("does not carry a held ending into the next phase", () => { + const outcome = makeAutonomousOutcome() + outcome.hold("max-steps") + outcome.record("pass", false, "model_error") + const next = outcome.record("close-out", true) + + assert.strictEqual(next.ending, undefined) + }) +}) diff --git a/apps/ai/src/chat/autonomous-outcome.ts b/apps/ai/src/chat/autonomous-outcome.ts new file mode 100644 index 0000000000..ddccd1af89 --- /dev/null +++ b/apps/ai/src/chat/autonomous-outcome.ts @@ -0,0 +1,76 @@ +/** + * How an autonomous turn's phases ended, and what the turn as a whole ends on. + * + * A pass, its restart or continuation, and its close-out are separate runs. Each one's ending is + * recorded here as it settles, so the turn's final `turn-end` and the reason the pull request is + * told are derived from every phase rather than from whichever variable the last one overwrote. + */ +import type { PrReviewFailureReason } from "@maple/domain/http" +import { Array as Arr, Option } from "effect" +import type { TurnEnding } from "./ChatSession" + +/** Which run of the turn a phase was. A restart starts the pass over, so it resets the failure. */ +export type PhaseKind = "pass" | "restart" | "continue" | "close-out" | "verdict" + +export interface PhaseOutcome { + readonly kind: PhaseKind + readonly submitted: boolean + /** The reason the run's own `turn-end` gave, held back while the turn decides what comes next. */ + readonly ending: TurnEnding["reason"] | undefined + /** Why the phase did not end in a report, when that is known. */ + readonly failure: PrReviewFailureReason | undefined +} + +/** A phase's own failure: one it raised, or a pass (not a close-out) that ran out of steps. */ +const phaseFailure = ( + kind: PhaseKind, + submitted: boolean, + ending: TurnEnding["reason"] | undefined, + raised: PrReviewFailureReason | undefined, +): PrReviewFailureReason | undefined => + raised ?? (!submitted && kind !== "close-out" && ending === "max-steps" ? "step_limit" : undefined) + +export const makeAutonomousOutcome = () => { + const phases: Array = [] + let held: TurnEnding["reason"] | undefined + + /** The phases since the latest restart: a restart discards the attempt it replaced. */ + const current = (): ReadonlyArray => { + const restart = Arr.findLastIndex(phases, (phase) => phase.kind === "restart") + return Option.match(restart, { onNone: () => phases, onSome: (index) => phases.slice(index) }) + } + + return { + /** The running phase's `turn-end`, held rather than written. */ + hold: (reason: TurnEnding["reason"]): void => { + held = reason + }, + /** Settle the running phase. `raised` is the failure it fell over with, if it did. */ + record: (kind: PhaseKind, submitted: boolean, raised?: PrReviewFailureReason): PhaseOutcome => { + const phase: PhaseOutcome = { + kind, + submitted, + ending: held, + failure: phaseFailure(kind, submitted, held, raised), + } + held = undefined + phases.push(phase) + return phase + }, + /** A failure decided between phases rather than by one, such as a pass that answered in prose. */ + fail: (failure: PrReviewFailureReason): void => { + phases.push({ kind: "verdict", submitted: false, ending: undefined, failure }) + }, + /** The first reason the current attempt stopped short, which is the one the pull request is told. */ + failure: (): PrReviewFailureReason | undefined => + Option.getOrUndefined(Arr.findFirst(current(), (phase) => Option.fromUndefinedOr(phase.failure))), + submitted: (): boolean => Arr.some(phases, (phase) => phase.submitted), + /** The turn's `turn-end`: a report that landed after any phase hit the step limit says so. */ + ending: (unsubmitted: string): TurnEnding => + Arr.some(phases, (phase) => phase.submitted) + ? { reason: Arr.some(phases, (phase) => phase.ending === "max-steps") ? "max-steps" : "stop" } + : { reason: "error", error: unsubmitted }, + } +} + +export type AutonomousOutcome = ReturnType diff --git a/apps/ai/src/chat/autonomous-settle.test.ts b/apps/ai/src/chat/autonomous-settle.test.ts new file mode 100644 index 0000000000..026f86d6fc --- /dev/null +++ b/apps/ai/src/chat/autonomous-settle.test.ts @@ -0,0 +1,123 @@ +import type { ChatAbortReason } from "@maple/domain/chat-session-stub" +import type { PrReviewFailureReason } from "@maple/domain/http" +import { Deferred, Effect, Fiber } from "effect" +import { assert, describe, it } from "vitest" +import { makeAutonomousOutcome } from "./autonomous-outcome" +import { NO_DIAGNOSIS_INTERRUPTED, makeUnfinishedSettler, runAutonomousTurn } from "./autonomous-settle" +import { interruptOnAbort, makeTurnObservability } from "./turn-outcome" + +/** A turn that owns all three rows, recording every write that marks one failed. */ +const harness = () => { + const controller = new AbortController() + const outcome = makeAutonomousOutcome() + const observability = makeTurnObservability() + const writes: Array = [] + const settler = makeUnfinishedSettler({ + outcome, + signal: controller.signal, + observability, + rows: { + failInvestigation: (error: string) => + Effect.sync(() => void writes.push(`investigation:${error}`)), + failReply: Effect.sync(() => void writes.push("reply")), + failReview: (reason: PrReviewFailureReason) => + Effect.sync(() => void writes.push(`review:${reason}`)), + }, + }) + let holds = true + const settle = settler.failUnfinished("no_report", "no diagnosis") + return { + outcome, + observability, + writes, + release: () => { + holds = false + }, + abort: (reason: ChatAbortReason) => controller.abort(reason), + /** The runner's wiring: the turn is interrupted when the session aborts it. */ + turn: (phases: Effect.Effect, settleWith: Effect.Effect = settle) => + runAutonomousTurn({ + phases, + settle: settleWith, + aborted: settler.aborted, + holdsTurn: () => holds, + }).pipe(interruptOnAbort(controller.signal, observability)), + } +} + +const INTERRUPTED_ROWS = [`investigation:${NO_DIAGNOSIS_INTERRUPTED}`, "reply", "review:interrupted"] + +describe("runAutonomousTurn", () => { + it("marks every row interrupted when the user aborts mid-phase", async () => { + const turn = harness() + const started = Deferred.makeUnsafe() + const fiber = Effect.runFork( + turn.turn(Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never))), + ) + await Effect.runPromise(Deferred.await(started)) + + turn.abort("user") + await Effect.runPromise(Fiber.await(fiber)) + + assert.deepEqual(turn.writes, INTERRUPTED_ROWS) + }) + + it("leaves the rows alone when a newer commit supersedes the review", async () => { + const turn = harness() + const started = Deferred.makeUnsafe() + const fiber = Effect.runFork( + turn.turn(Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never))), + ) + await Effect.runPromise(Deferred.await(started)) + + turn.abort("superseded") + await Effect.runPromise(Fiber.await(fiber)) + + assert.deepEqual(turn.writes, []) + }) + + it("marks the rows interrupted when the slot expired between phases", async () => { + const turn = harness() + + await Effect.runPromise(turn.turn(Effect.sync(() => turn.release()))) + + assert.deepEqual(turn.writes, INTERRUPTED_ROWS) + assert.strictEqual(turn.observability.outcome, "aborted") + }) + + it("settles a turn that still holds its slot with its own failure, not as interrupted", async () => { + const turn = harness() + + await Effect.runPromise(turn.turn(Effect.void)) + + assert.deepEqual(turn.writes, ["investigation:no diagnosis", "reply", "review:no_report"]) + assert.strictEqual(turn.observability.outcome, undefined) + }) + + it("writes nothing for a turn whose report landed before the abort", async () => { + const turn = harness() + turn.outcome.record("pass", true) + + await Effect.runPromise(turn.turn(Effect.sync(() => turn.release()))) + + assert.deepEqual(turn.writes, []) + }) + + it("finishes settling once when an abort arrives mid-settle", async () => { + const turn = harness() + const settling = Deferred.makeUnsafe() + const proceed = Deferred.makeUnsafe() + const settle = Deferred.succeed(settling, undefined).pipe( + Effect.andThen(Deferred.await(proceed)), + Effect.andThen(Effect.sync(() => void turn.writes.push("settled"))), + ) + const fiber = Effect.runFork(turn.turn(Effect.void, settle)) + await Effect.runPromise(Deferred.await(settling)) + + turn.abort("user") + await Effect.runPromise(Deferred.succeed(proceed, undefined)) + await Effect.runPromise(Fiber.await(fiber)) + + assert.deepEqual(turn.writes, ["settled"]) + }) +}) diff --git a/apps/ai/src/chat/autonomous-settle.ts b/apps/ai/src/chat/autonomous-settle.ts new file mode 100644 index 0000000000..5c9d78b5c9 --- /dev/null +++ b/apps/ai/src/chat/autonomous-settle.ts @@ -0,0 +1,83 @@ +/** + * How an autonomous turn's rows are settled when it stops without a report: after its phases, or + * when an abort or a stale expiry cuts them short. The runner supplies the phases and the writes. + */ +import type { PrReviewFailureReason } from "@maple/domain/http" +import { Effect, Option } from "effect" +import type { AutonomousOutcome } from "./autonomous-outcome" +import { abortReason, type TurnObservability } from "./turn-outcome" + +/** What the investigation row records when the turn was stopped before its report. */ +export const NO_DIAGNOSIS_INTERRUPTED = + "interrupted: the run was stopped before it submitted a diagnosis; retry" +/** How long a row write that settles an unfinished turn may take; it can run in a finalizer. */ +const ROW_WRITE_TIMEOUT = "10 seconds" + +/** The rows this turn owns, each with the write that marks it failed. */ +export interface UnfinishedRows { + readonly failInvestigation?: (error: string) => Effect.Effect + readonly failReply?: Effect.Effect + readonly failReview?: (reason: PrReviewFailureReason) => Effect.Effect +} + +const settleRow = (write: Effect.Effect, failed: string) => + write.pipe( + Effect.timeout(ROW_WRITE_TIMEOUT), + Effect.catchCause((cause) => Effect.logError(failed, cause)), + ) + +export const makeUnfinishedSettler = (options: { + readonly outcome: AutonomousOutcome + readonly signal: AbortSignal | undefined + readonly observability: TurnObservability + readonly rows: UnfinishedRows +}) => { + const { outcome, signal, observability, rows } = options + /** Not on a supersede: the newer head's review owns the row and marks it skipped itself. */ + const failUnfinished = (reason: PrReviewFailureReason, investigationError: string) => + Effect.gen(function* () { + if (outcome.submitted()) return + if (Option.contains(abortReason(signal), "superseded")) return + if (rows.failInvestigation !== undefined) { + observability.failureReason = "NoDiagnosis" + yield* settleRow( + rows.failInvestigation(investigationError), + "Could not record the failed pass", + ) + } + if (rows.failReply !== undefined) { + observability.failureReason = "NoReply" + yield* settleRow(rows.failReply, "Could not record the failed reply") + } + if (rows.failReview !== undefined) { + observability.failureReason = "NoReview" + yield* Effect.annotateCurrentSpan("maple.pr_review.failure_reason", reason) + yield* settleRow(rows.failReview(reason), "Could not record the failed review") + } + }) + + /** An abort between or during phases still settles the rows: nothing else moves them off `running`. */ + const aborted = Effect.suspend(() => { + observability.outcome = "aborted" + return failUnfinished("interrupted", NO_DIAGNOSIS_INTERRUPTED) + }) + + return { failUnfinished, aborted } +} + +/** + * Run the phases, then settle exactly once: `settle` while the turn still holds its slot, else + * `aborted`. Settling is uninterruptible, so an abort arriving then cannot settle the rows twice. + */ +export const runAutonomousTurn = (options: { + readonly phases: Effect.Effect + readonly settle: Effect.Effect + readonly aborted: Effect.Effect + readonly holdsTurn: () => boolean +}): Effect.Effect => + Effect.uninterruptibleMask((restore) => + restore(options.phases).pipe( + Effect.onInterrupt(() => options.aborted), + Effect.andThen(Effect.suspend(() => (options.holdsTurn() ? options.settle : options.aborted))), + ), + ) diff --git a/apps/ai/src/chat/progress.test.ts b/apps/ai/src/chat/progress.test.ts index fb7281fdbc..e42ffbfe90 100644 --- a/apps/ai/src/chat/progress.test.ts +++ b/apps/ai/src/chat/progress.test.ts @@ -136,6 +136,31 @@ describe("makeProgressFeed", () => { }), ) + /** A report refused after the turn closed the feed must not reopen a feed with no writer. */ + it.effect("ignores a resume once the feed is closed", () => + Effect.gen(function* () { + const { feed, writes } = yield* feedWith() + yield* feed.close + yield* feed.resume + feed.step("search_logs", {}) + yield* feed.flush + yield* TestClock.adjust(Duration.times(INVESTIGATION_PROGRESS_HEARTBEAT, 2)) + assert.lengthOf(writes, 0) + }), + ) + + it.effect("stays closed when a pause follows the close", () => + Effect.gen(function* () { + const { feed, writes } = yield* feedWith() + yield* feed.close + yield* feed.pause + yield* feed.resume + feed.step("search_logs", {}) + yield* feed.flush + assert.lengthOf(writes, 0) + }), + ) + it.effect("gives up on a hung write instead of holding the permit", () => Effect.gen(function* () { const feed = yield* makeProgressFeed({ diff --git a/apps/ai/src/chat/progress.ts b/apps/ai/src/chat/progress.ts index 6f6fa20e72..7244c5dac2 100644 --- a/apps/ai/src/chat/progress.ts +++ b/apps/ai/src/chat/progress.ts @@ -118,9 +118,9 @@ export interface ProgressFeed { readonly flush: Effect.Effect /** Stop writing, waiting out a write in flight. Steps still queue, for a `resume`. */ readonly pause: Effect.Effect - /** Write again after a `pause`, as when the report it made way for was refused. */ + /** Write again after a `pause`, as when the report it made way for was refused. A no-op once closed. */ readonly resume: Effect.Effect - /** Stop the feed for good, waiting out a write in flight. Safe to call twice. */ + /** Stop the feed for good, waiting out a write in flight. Safe to call twice; final. */ readonly close: Effect.Effect } @@ -130,6 +130,9 @@ interface ToolCall { readonly at: number } +/** Whether the feed writes. `Closed` is final: its writer fiber is gone, so nothing resumes it. */ +type FeedPhase = "Open" | "Paused" | "Closed" + interface FeedState { readonly stepCount: number readonly steps: ReadonlyArray @@ -150,7 +153,7 @@ export const makeProgressFeed = Effect.fnUntraced(function* (options: { const calls = yield* Queue.unbounded() const state = yield* Ref.make({ stepCount: 0, steps: [], dirty: false }) const permit = yield* Semaphore.make(1) - const open = yield* Ref.make(true) + const phase = yield* Ref.make("Open") const absorb = (batch: ReadonlyArray) => Ref.update(state, (current) => @@ -174,7 +177,7 @@ export const makeProgressFeed = Effect.fnUntraced(function* (options: { // timeout, so a hung request cannot hold the permit (and the report behind it) for long. const write = Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - if (!(yield* Ref.get(open))) return + if ((yield* Ref.get(phase)) !== "Open") return yield* Queue.clear(calls).pipe(Effect.flatMap(absorb)) const current = yield* Ref.get(state) if (!current.dirty) return @@ -206,17 +209,20 @@ export const makeProgressFeed = Effect.fnUntraced(function* (options: { Effect.all([first, beat], { concurrency: "unbounded", discard: true }), ) - // The flag first, so a write still queued for the permit gives way at once; then the permit, + // The phase first, so a write still queued for the permit gives way at once; then the permit, // so the one write in flight (bounded by its timeout) finishes before anything follows. - const pause = Ref.set(open, false).pipe(Effect.andThen(permit.withPermits(1)(Effect.void))) + const stop = (next: Exclude) => + Ref.update(phase, (current) => (current === "Closed" ? current : next)).pipe( + Effect.andThen(permit.withPermits(1)(Effect.void)), + ) return { step: (tool, input) => { Queue.offerUnsafe(calls, { tool, input, at: clock.currentTimeMillisUnsafe() }) }, flush: write, - pause, - resume: Ref.set(open, true), - close: pause.pipe(Effect.andThen(Fiber.interrupt(fiber))), + pause: stop("Paused"), + resume: Ref.update(phase, (current) => (current === "Paused" ? "Open" : current)), + close: stop("Closed").pipe(Effect.andThen(Fiber.interrupt(fiber))), } satisfies ProgressFeed }) diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index e44fd37b52..c899dd86db 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -31,7 +31,6 @@ import { Clock, Duration, Effect, - Exit, Layer, ManagedRuntime, Match, @@ -45,7 +44,6 @@ import { FetchHttpClient } from "effect/http" import type { WorkersAiBinding } from "../platform/WorkersAiHttpClient" import { ReturnedToolFailuresOkLive } from "../platform/genai-spans" import type { ChatSession } from "./ChatSession" -import type { ChatTurnEvent } from "./events" import { withToolTranscript } from "./close-out" import { CLOSE_OUT_PROMPT, @@ -71,9 +69,11 @@ import { } from "./tools" import { interruptOnAbort, makeTurnObservability, turnSpanAttributes } from "./turn-outcome" +import { makeAutonomousOutcome, type PhaseKind } from "./autonomous-outcome" +import { makeUnfinishedSettler, runAutonomousTurn } from "./autonomous-settle" import { type AgentDefinition, agentForSession } from "./agents" import { profileForTurn } from "./profiles" -import { runChatTurn, type ChatRunOutcome } from "./run" +import { runChatTurn } from "./run" import { toTenantContext, turnTenant } from "./turn-actor" import { summarizeCause } from "@maple/backend/platform/describe-cause" import { trackAiCredits, trackTokenUsage } from "@maple/backend/services/billing/autumn-tracker" @@ -140,8 +140,8 @@ export interface RunChatSessionTurnInput { /** * Metering is housekeeping, and it runs after the answer, on the way out of the turn. * - * `endTurn` sits in the `finally` of `ChatSession.runTurn`, so whatever the metering finalizer - * waits on holds the session's turn slot, and the stale-claim reclaim is tens of minutes out + * A turn that did not end itself is ended in the `finally` of `ChatSession.runTurn`, so whatever + * the metering finalizer waits on can hold the session's turn slot, and the stale-claim reclaim is tens of minutes out * (`TURN_STALE_MS` in `ChatSession.ts`). Unbounded, a POST to Autumn that never answers is a way * for a third party's outage to wedge a conversation. * @@ -150,14 +150,14 @@ export interface RunChatSessionTurnInput { */ const METERING_TIMEOUT = "5 seconds" -/** Stable copy for the durable/browser event; detailed causes stay server-side. */ -const CHAT_TURN_FAILED = "Maple couldn't complete this response." const NO_DIAGNOSIS_MESSAGE = "Maple ended this investigation without a diagnosis." /** What the row records when the pass and its close-out both ended in prose or on an error. */ const NO_DIAGNOSIS_ERROR = "no_diagnosis: the agent ended its pass without submitting a diagnosis; retry" const NO_REVIEW_MESSAGE = "Maple ended this review without a report." const NO_REPLY_MESSAGE = "Maple ended this answer without posting it." const NO_REPLY_ERROR = "no_reply: the agent ended its pass without submitting an answer" +/** How long filing a review's saved findings may take once the turn has started settling. */ +const SUBMIT_SAVED_TIMEOUT = "60 seconds" /** A review pass that answered in prose gets another pass only with at least this much of its clock left. */ const MIN_CONTINUE_MS = 2 * 60 * 1000 @@ -331,7 +331,6 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis // so there is no later moment to hand it a total — and the metering finalizer reads it after the // turn has ended, including when it ended by failing. const usage = makeRunUsage() - let recordedTerminal = false // `empty_output` and `recovery_count` are gone with the machinery that produced them: they // counted the old loop's own repair attempts, and the engine has no equivalent to report. const annotateTurn = () => Effect.annotateCurrentSpan(turnSpanAttributes(observability, input.signal)) @@ -355,13 +354,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis Effect.sync(() => { observability.outcome = "error" observability.failureReason = `LinkedUserNotPermitted:${refusal.reason}` - input.session.append({ - type: "turn-end", - messageId: input.messageId, - reason: "error", - error: refusal.message, - }) - recordedTerminal = true + input.session.endTurn(input.messageId, { reason: "error", error: refusal.message }) return undefined }), ), @@ -400,8 +393,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }) if (settled) { observability.outcome = "resume_skipped" - input.session.append({ type: "turn-end", messageId: input.messageId, reason: "stop" }) - recordedTerminal = true + input.session.endTurn(input.messageId, { reason: "stop" }) yield* annotateTurn() return } @@ -558,15 +550,12 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis investigationFeed === undefined ? Effect.void : investigationFeed.flush.pipe(Effect.andThen(investigationFeed.close)) - // Why the pass, or else its close-out, stopped; the first reason is the one the PR is told. - let failure: PrReviewFailureReason | undefined + // Every phase's ending, recorded as it settles: the PR's reason and the final turn-end + // are read from all of them. + const outcome = makeAutonomousOutcome() const holdsTurn = () => input.session.holdsTurn(input.messageId) - // An autonomous pass ends when the runner says so, not when a run does: a run that stopped - // in prose or died on a model error gets one close-out turn first, so its terminal is held - // back until the outcome is known. - let held: Extract | undefined - // Read through a function: `held` is set by the run's callback, which narrowing cannot see. - const ranOutOfSteps = () => held?.reason === "max-steps" + // Whether a conversation turn's run wrote its own ending (non-autonomous only). + let recordedTerminal = false const run = (turn: { readonly text: string readonly history: ReadonlyArray @@ -589,9 +578,13 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis reviews.submitReview( orgId, reviewId, - request.partial === true && failure !== undefined - ? new SubmitPrReviewRequest({ ...request, partialReason: failure }) - : request, + Option.match(Option.fromUndefinedOr(outcome.failure()), { + onNone: () => request, + onSome: (partialReason) => + request.partial === true + ? new SubmitPrReviewRequest({ ...request, partialReason }) + : request, + }), ), ), submitReply: conversations.submitReply, @@ -622,40 +615,46 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis } if (event.type === "turn-end" && event.task === undefined) { observability.outcome = event.reason - if (autonomous && event.reason !== "aborted") { - held = event - return - } - recordedTerminal = true + // An autonomous pass ends when the runner says so, not when a run does. + if (autonomous && event.reason !== "aborted") return outcome.hold(event.reason) + const { reason, error } = event + recordedTerminal = input.session.endTurn( + input.messageId, + error === undefined ? { reason } : { reason, error }, + ) + return } input.session.append(event) }, // oxlint-disable-next-line effecttsgo/strict-effect-provide -- the turn's OpenRouter client is built per turn from its settings }).pipe(Effect.provide(openRouterClient)) - // A pass that failed is a pass with no diagnosis yet, not a dead turn: the close-out below - // still gets its say. Interrupts stay interrupts. - const recoverAutonomousFailure = (effect: Effect.Effect) => - effect.pipe( + /** + * One phase, settled into `outcome`. A pass that failed is a pass with no report yet, not a + * dead turn: the close-out still gets its say. Interrupts stay interrupts. + */ + const phase = (kind: PhaseKind, turn: Parameters[0]) => + run(turn).pipe( + Effect.map((result) => ({ result, raised: undefined })), Effect.catchCauseIf( (cause) => !Cause.hasInterruptsOnly(cause), (cause) => - Effect.sync(() => { - failure ??= reviewFailureReason(cause, { ownKey: Option.isSome(orgKey) }) - }).pipe( - Effect.andThen(Effect.logWarning("Autonomous pass failed; closing it out")), + Effect.logWarning("Autonomous pass failed; closing it out").pipe( Effect.annotateLogs({ sessionId: input.sessionId, messageId: input.messageId, cause: summarizeCause(cause), }), - Effect.as({ - autonomous: true, - submitted: false, - compactions: 0, + Effect.as({ + result: { autonomous: true, submitted: false, compactions: 0 }, + raised: reviewFailureReason(cause, { ownKey: Option.isSome(orgKey) }), }), ), ), + Effect.map(({ result, raised }) => { + observability.compactions = (observability.compactions ?? 0) + result.compactions + return outcome.record(kind, result.submitted, raised) + }), ) if (!autonomous) { @@ -665,65 +664,79 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis return } - const passDeadline = - (yield* Clock.currentTimeMillis) + - Duration.toMillis(Duration.fromInputUnsafe(agent.budget.maxDuration)) - const first = yield* recoverAutonomousFailure(run({ text, history: prior })) - observability.compactions = first.compactions - let submitted = first.submitted - // Exhausted rather than failed: the engine's final answer ended in prose, not a report. - if (!submitted && held?.reason === "max-steps") failure ??= "step_limit" - // A review stopped for failing tool calls is usually a bad start, not an impossible review: it - // starts over once from the kickoff, inside the same wall clock, without the failed calls. - if (!submitted && failure === "stuck" && prReviewId !== undefined && holdsTurn()) { - const remaining = passDeadline - (yield* Clock.currentTimeMillis) - if (remaining >= MIN_CONTINUE_MS) { - failure = undefined - held = undefined - const restarted = yield* recoverAutonomousFailure( - run({ + const { failUnfinished, aborted } = makeUnfinishedSettler({ + outcome, + signal: input.signal, + observability, + rows: { + ...(investigationId === undefined + ? undefined + : { + failInvestigation: (error: string) => + investigations.failInvestigation(tenant.orgId, investigationId, error), + }), + ...(prReplyId === undefined + ? undefined + : { failReply: conversations.failReply(tenant.orgId, prReplyId, NO_REPLY_ERROR) }), + ...(prReviewId === undefined + ? undefined + : { + failReview: (reason: PrReviewFailureReason) => + reviews.failReview(tenant.orgId, prReviewId, reviewFailureError(reason)), + }), + }, + }) + + const autonomousTurn = Effect.gen(function* () { + const passDeadline = + (yield* Clock.currentTimeMillis) + + Duration.toMillis(Duration.fromInputUnsafe(agent.budget.maxDuration)) + const remainingAgent = (remaining: number): AgentDefinition => ({ + ...agent, + budget: { ...agent.budget, maxDuration: Duration.millis(remaining) }, + }) + yield* phase("pass", { text, history: prior }) + // A review stopped for failing tool calls is usually a bad start, not an impossible review: + // it starts over once from the kickoff, inside the same wall clock, without the failed calls. + if ( + !outcome.submitted() && + outcome.failure() === "stuck" && + prReviewId !== undefined && + holdsTurn() + ) { + const remaining = passDeadline - (yield* Clock.currentTimeMillis) + if (remaining >= MIN_CONTINUE_MS) { + const restarted = yield* phase("restart", { text, history: prior, - agent: { - ...agent, - budget: { ...agent.budget, maxDuration: Duration.millis(remaining) }, - }, - }), - ) - observability.compactions = (observability.compactions ?? 0) + restarted.compactions - submitted = restarted.submitted - yield* Effect.annotateCurrentSpan("maple.pr_review.restarted", submitted) - if (!submitted && ranOutOfSteps()) failure ??= "step_limit" + agent: remainingAgent(remaining), + }) + yield* Effect.annotateCurrentSpan("maple.pr_review.restarted", restarted.submitted) + } } - } - // A review that answered in prose with no limit hit was not stopped, it stopped itself. With - // time left it gets one more pass that may still read, inside the same wall clock. - if (!submitted && failure === undefined && prReviewId !== undefined && holdsTurn()) { - const remaining = passDeadline - (yield* Clock.currentTimeMillis) - if (remaining >= MIN_CONTINUE_MS) { - held = undefined - const continued = yield* recoverAutonomousFailure( - run({ + // A review that answered in prose with no limit hit was not stopped, it stopped itself. With + // time left it gets one more pass that may still read, inside the same wall clock. + if ( + !outcome.submitted() && + outcome.failure() === undefined && + prReviewId !== undefined && + holdsTurn() + ) { + const remaining = passDeadline - (yield* Clock.currentTimeMillis) + if (remaining >= MIN_CONTINUE_MS) { + const continued = yield* phase("continue", { // The replayed history keeps only its newest part, which can drop the kickoff. text: `${PR_REVIEW_CONTINUE_PROMPT}\n\n${text}`, history: withToolTranscript(input.session.history()), - agent: { - ...agent, - budget: { ...agent.budget, maxDuration: Duration.millis(remaining) }, - }, - }), - ) - observability.compactions = (observability.compactions ?? 0) + continued.compactions - submitted = continued.submitted - yield* Effect.annotateCurrentSpan("maple.pr_review.continued", submitted) - if (!submitted && ranOutOfSteps()) failure ??= "step_limit" + agent: remainingAgent(remaining), + }) + yield* Effect.annotateCurrentSpan("maple.pr_review.continued", continued.submitted) + } + if (!outcome.submitted() && outcome.failure() === undefined) + outcome.fail("answered_in_prose") } - if (!submitted) failure ??= "answered_in_prose" - } - if (!submitted && holdsTurn()) { - held = undefined - const closeOut = yield* recoverAutonomousFailure( - run({ + if (!outcome.submitted() && holdsTurn()) { + const closeOut = yield* phase("close-out", { text: prReplyId !== undefined ? PR_REPLY_CLOSE_OUT_PROMPT @@ -732,51 +745,36 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis : PR_REVIEW_CLOSE_OUT_PROMPT, history: withToolTranscript(input.session.history()), closeOut: true, - }), - ) - observability.compactions = (observability.compactions ?? 0) + closeOut.compactions - submitted = closeOut.submitted - yield* Effect.annotateCurrentSpan( - prReplyId !== undefined - ? "maple.pr_reply.closed_out" - : prReviewId === undefined - ? "maple.investigation.closed_out" - : "maple.pr_review.closed_out", - submitted, - ) - } + }) + yield* Effect.annotateCurrentSpan( + prReplyId !== undefined + ? "maple.pr_reply.closed_out" + : prReviewId === undefined + ? "maple.investigation.closed_out" + : "maple.pr_review.closed_out", + closeOut.submitted, + ) + } - yield* closeFeeds + yield* closeFeeds + }) - if (holdsTurn()) { - if (!submitted && investigationId !== undefined) { - observability.failureReason = "NoDiagnosis" - yield* investigations - .failInvestigation(tenant.orgId, investigationId, NO_DIAGNOSIS_ERROR) - .pipe( - Effect.catchCause((cause) => - Effect.logError("Could not record the failed pass", cause), - ), - ) - } - if (!submitted && prReplyId !== undefined) { - observability.failureReason = "NoReply" - yield* conversations - .failReply(tenant.orgId, prReplyId, NO_REPLY_ERROR) - .pipe( - Effect.catchCause((cause) => - Effect.logError("Could not record the failed reply", cause), - ), - ) - } + // A turn that still holds its slot after the phases settles here, once (`runAutonomousTurn`). + const settleTurn = Effect.gen(function* () { // Findings the pass saved, and its children's warnings, are posted as a partial rather // than lost with the report. const saved = closeOutFindings( review?.ledger.findings() ?? [], review?.childFindings?.all() ?? [], ) - if (!submitted && prReviewId !== undefined && review !== undefined && saved.length > 0) { - submitted = yield* reviews + if ( + !outcome.submitted() && + prReviewId !== undefined && + review !== undefined && + saved.length > 0 + ) { + const failure = outcome.failure() + const filed = yield* reviews .submitReview( tenant.orgId, prReviewId, @@ -790,51 +788,31 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis ) .pipe( Effect.as(true), + // Settling is uninterruptible, so a hung submit is bounded; `failReview` skips a filed review. + Effect.timeout(SUBMIT_SAVED_TIMEOUT), Effect.catchCause((cause) => Effect.logError("Could not file the saved findings", cause).pipe( Effect.as(false), ), ), ) - yield* Effect.annotateCurrentSpan("maple.pr_review.filed_saved_findings", submitted) + outcome.record("verdict", filed) + yield* Effect.annotateCurrentSpan("maple.pr_review.filed_saved_findings", filed) } - if (!submitted && prReviewId !== undefined) { - observability.failureReason = "NoReview" - yield* Effect.annotateCurrentSpan( - "maple.pr_review.failure_reason", - failure ?? "no_report", - ) - yield* reviews - .failReview(tenant.orgId, prReviewId, reviewFailureError(failure ?? "no_report")) - .pipe( - Effect.catchCause((cause) => - Effect.logError("Could not record the failed review", cause), - ), - ) - } - input.session.append( - submitted - ? { - type: "turn-end", - messageId: input.messageId, - reason: held?.reason === "max-steps" ? "max-steps" : "stop", - } - : { - type: "turn-end", - messageId: input.messageId, - reason: "error", - error: - prReplyId !== undefined - ? NO_REPLY_MESSAGE - : prReviewId === undefined - ? NO_DIAGNOSIS_MESSAGE - : NO_REVIEW_MESSAGE, - }, + yield* failUnfinished(outcome.failure() ?? "no_report", NO_DIAGNOSIS_ERROR) + input.session.endTurn( + input.messageId, + outcome.ending( + prReplyId !== undefined + ? NO_REPLY_MESSAGE + : prReviewId === undefined + ? NO_DIAGNOSIS_MESSAGE + : NO_REVIEW_MESSAGE, + ), ) - recordedTerminal = true - } else { - observability.outcome = "aborted" - } + }) + + yield* runAutonomousTurn({ phases: autonomousTurn, settle: settleTurn, aborted, holdsTurn }) yield* annotateTurn() }) @@ -899,17 +877,9 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }), ) - const exit = await runtime.runPromiseExit(program) - // The detailed cause belongs in server logs and the failed Effect span, never in the durable - // event the browser reads back. - if (Exit.isFailure(exit) && input.session.holdsTurn(input.messageId)) { - input.session.append({ - type: "turn-end", - messageId: input.messageId, - reason: "error", - error: CHAT_TURN_FAILED, - }) - } + // A turn that failed is ended by the session once this returns (`ChatSession.runTurn`), with + // stable copy: the detailed cause stays in server logs and the failed span. + await runtime.runPromiseExit(program) await Effect.runPromise(releaseQuietly(runtime, () => telemetry.flush(input.env))) } diff --git a/apps/chat-bot/src/relay/ConnectorRelay.test.ts b/apps/chat-bot/src/relay/ConnectorRelay.test.ts index 8930aa44f6..6c268e1347 100644 --- a/apps/chat-bot/src/relay/ConnectorRelay.test.ts +++ b/apps/chat-bot/src/relay/ConnectorRelay.test.ts @@ -275,6 +275,28 @@ describe("turn jobs", () => { expect([...state.turnJobs]).toEqual([]) }) + it("keeps a handed-off turn's checkpoint, and settles it on its next job", async () => { + const state = objectState() + const run = heavy({ + settle: "done", + event: async (host) => { + await Effect.runPromise(host.recordTurn(checkpoint())) + await Effect.runPromise(host.handOffTurn(checkpoint())) + }, + }) + const relay = relayOn(state, run.load) + + await Effect.runPromise(relay.deliver(message)) + await Promise.all(state.pending) + // The run ended, but the turn did not: its checkpoint and job survive it. + expect([...state.stored.keys()]).toEqual([TURN_KEY]) + expect(state.turnJobs.get(TURN_KEY)).toBe(1) + + await Effect.runPromise(relay.turnDue(TURN_KEY)) + expect(run.settled).toEqual([checkpoint()]) + expect([...state.stored]).toEqual([]) + }) + it("settles an evicted turn, and drops it and its job once settled", async () => { const state = objectState() const run = heavy({ settle: "done" }) diff --git a/apps/chat-bot/src/relay/ConnectorRelay.ts b/apps/chat-bot/src/relay/ConnectorRelay.ts index 37b6082ef1..b830358a0f 100644 --- a/apps/chat-bot/src/relay/ConnectorRelay.ts +++ b/apps/chat-bot/src/relay/ConnectorRelay.ts @@ -317,10 +317,23 @@ export class ConnectorRelay { recorded = turnKey(checkpoint) return this.recordTurn(checkpoint) }) + // A handed-off turn keeps its checkpoint and leaves `relaying`, so its next job settles it. + const handOffTurn = (checkpoint: RelayTurnCheckpoint) => + Effect.sync(() => { + const key = turnKey(checkpoint) + if (recorded === key) recorded = undefined + this.relaying.delete(key) + }) return runtimeCall("event", async () => { const { runInboundEvent } = await this.runtime() await runInboundEvent( - { env: this.env, chatSessions: this.chatSessions, ...this.relayPorts(event), recordTurn }, + { + env: this.env, + chatSessions: this.chatSessions, + ...this.relayPorts(event), + recordTurn, + handOffTurn, + }, event, ) }).pipe( diff --git a/apps/chat-bot/src/relay/run.ts b/apps/chat-bot/src/relay/run.ts index f7f2dddb98..9431e6326c 100644 --- a/apps/chat-bot/src/relay/run.ts +++ b/apps/chat-bot/src/relay/run.ts @@ -119,6 +119,8 @@ export interface RelayHost { readonly rememberConversation: RelayPorts["rememberConversation"] /** The relay object's checkpoint of the turn it is rendering — see `./settle.ts`. */ readonly recordTurn: RelayPorts["recordTurn"] + /** The relay object's release of a turn it stopped following, checkpoint kept. */ + readonly handOffTurn: RelayPorts["handOffTurn"] } /** @@ -327,6 +329,7 @@ const ports = ( ownsConversation: host.ownsConversation, rememberConversation: host.rememberConversation, recordTurn: host.recordTurn, + handOffTurn: host.handOffTurn, alertAck: alertAck(host), }) @@ -403,6 +406,8 @@ export const settleInboundTurn = async ( announceUnlinked: Effect.succeed(false), ownsConversation: () => Effect.succeed(false), rememberConversation: () => Effect.void, + // A settle renders from a finished log, so it never outlives its timeout to hand anything off. + handOffTurn: () => Effect.void, } return Effect.runPromise( Effect.gen(function* () { diff --git a/apps/chat-bot/src/relay/settle.test.ts b/apps/chat-bot/src/relay/settle.test.ts index ddb24fb214..d695e943de 100644 --- a/apps/chat-bot/src/relay/settle.test.ts +++ b/apps/chat-bot/src/relay/settle.test.ts @@ -187,6 +187,7 @@ const ports = ( stored.writes += 1 return onRecord(checkpoint) }), + handOffTurn: () => Effect.void, }) /** What a fresh activation reads back — through the schema, as the object does. */ @@ -356,7 +357,7 @@ describe("settling a relayed turn", () => { it.effect("drops a checkpoint older than any turn the session would still be running", () => Effect.gen(function* () { const chat = platform() - yield* TestClock.adjust("31 minutes") + yield* TestClock.adjust("36 minutes") const outcome = yield* settleRelayedTurn( checkpointOf(["m1"]), diff --git a/apps/chat-bot/src/relay/settle.ts b/apps/chat-bot/src/relay/settle.ts index 5a6c641e1b..689f3bf4eb 100644 --- a/apps/chat-bot/src/relay/settle.ts +++ b/apps/chat-bot/src/relay/settle.ts @@ -35,8 +35,8 @@ export type RelayTurnCheckpoint = typeof RelayTurnCheckpoint.Type export const decodeRelayTurnCheckpoint = Schema.decodeUnknownOption(RelayTurnCheckpoint) -/** The session's `TURN_STALE_MS` (25 min, when it expires a turn nobody ended) plus a margin. */ -const CHECKPOINT_TTL_MS = 30 * 60 * 1000 +/** Past the session's `TURN_STALE_MS` (25 min) and the relay's timeout (27 min), so a handed-off turn still settles. */ +const CHECKPOINT_TTL_MS = 35 * 60 * 1000 /** `"pending"`: the session is still running the turn, so the checkpoint waits for its next job. */ export type SettleOutcome = "pending" | "done" diff --git a/apps/chat-bot/src/relay/turn.test.ts b/apps/chat-bot/src/relay/turn.test.ts index 9f501383ce..e96ce7980e 100644 --- a/apps/chat-bot/src/relay/turn.test.ts +++ b/apps/chat-bot/src/relay/turn.test.ts @@ -43,6 +43,7 @@ import { Context, Duration, Effect, + Fiber, Logger, Option, References, @@ -54,6 +55,7 @@ import { import { TestClock } from "effect/testing" import { ConversationNotRecorded, FOLLOW_UP_WINDOW_MS } from "./conversation.ts" import { + RELAY_TIMEOUT, relayInboundEvent, relayWithWorkspace, WorkspaceLookupFailed, @@ -369,6 +371,7 @@ const host = ( ) : Effect.sync(() => void opened.add(conversation.conversationKey)), recordTurn: () => Effect.void, + handOffTurn: () => Effect.void, alertAck: (request) => Effect.sync(() => { alertAcks.push(request) @@ -486,6 +489,64 @@ describe("relaying a mention", () => { }), ) + it.effect("leaves the streaming render and keeps the checkpoint when it stops following a turn", () => + Effect.gen(function* () { + const platform = chat() + const agent = session([]) + // A connection that never closes: the session is still running the turn at the timeout. + const frames = new TextEncoder().encode( + "retry: 1000\n\n" + + Arr.join( + Arr.map( + [ + event(1, { type: "turn-start", messageId: "a1" }), + event(2, { type: "text-delta", messageId: "a1", text: "Still digging" }), + ], + (next) => `id: ${next.seq}\ndata: ${JSON.stringify(next)}\n\n`, + ), + "", + ), + ) + const stub: ChatSessionStub = { + ...agent.stub, + subscribe: () => + Promise.resolve( + new ReadableStream({ start: (controller) => controller.enqueue(frames) }), + ), + } + const deployment = host(platform.outbound, stub) + const recorded: Array = [] + const handedOff: Array = [] + const ports: RelayPorts = { + ...deployment.ports, + recordTurn: (checkpoint) => Effect.sync(() => void recorded.push(checkpoint.turnMessageId)), + handOffTurn: (checkpoint) => Effect.sync(() => void handedOff.push(checkpoint.turnMessageId)), + } + + const relayed = yield* Effect.forkChild(relayInboundEvent(mention, ports)) + // The stream is read off real promises: step both clocks until the streamed text is shown. + const shown = () => Arr.some(platform.calls, (call) => prose(call.blocks) === "Still digging") + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 1))).pipe( + Effect.andThen(TestClock.adjust("10 millis")), + Effect.repeat({ until: shown, times: 500 }), + ) + expect(shown()).toBe(true) + yield* TestClock.adjust(RELAY_TIMEOUT) + yield* Fiber.join(relayed) + + // The checkpoint the driver recorded is handed to the settle path, not dropped. + expect(recorded).toContain("a1") + expect(handedOff).toEqual(["a1"]) + // And the message leaves "Working on it" for what was said, with a line saying more is coming. + const last = platform.calls[platform.calls.length - 1] + expect(last?.verb).toBe("edit") + expect(prose(blocksOf(platform.calls))).toBe("Still digging") + expect(notices(blocksOf(platform.calls))).toEqual([ + "Still working on this. The full answer will replace this message when it is ready.", + ]) + }), + ) + it.effect("relays a chart the agent drew as the picture this deployment can sign", () => Effect.gen(function* () { const platform = chat() diff --git a/apps/chat-bot/src/relay/turn.ts b/apps/chat-bot/src/relay/turn.ts index 46667a4709..03645f1c29 100644 --- a/apps/chat-bot/src/relay/turn.ts +++ b/apps/chat-bot/src/relay/turn.ts @@ -150,18 +150,20 @@ export interface RelayPorts { readonly announceUnlinked: Effect.Effect /** Checkpoint the turn's posted messages, for settling it after an eviction (`./settle.ts`). */ readonly recordTurn: (checkpoint: RelayTurnCheckpoint) => Effect.Effect + /** + * Stop following a turn that outlived {@link RELAY_TIMEOUT}, keeping its checkpoint: the + * scheduled settle (`./settle.ts`) renders the answer once the session ends the turn. + */ + readonly handOffTurn: (checkpoint: RelayTurnCheckpoint) => Effect.Effect /** Acknowledge or unacknowledge an alert incident from its card. Never fails; see the outcome. */ readonly alertAck: (request: RelayAlertAckRequest) => Effect.Effect } /** - * How long a relayed turn may run before it is abandoned. - * - * The session ends a turn of its own accord — the run finishes, or its heartbeat fails a turn - * whose object was evicted — so this is the bound on everything that could keep the stream open - * without one, and it matches the session's own staleness ceiling (`TURN_STALE_MS`). + * How long a relayed turn is followed before it is handed to the settle path: two minutes past the + * session's staleness ceiling (`TURN_STALE_MS`, 25 min), so the session's own ending wins the race. */ -const RELAY_TIMEOUT = Duration.minutes(25) +export const RELAY_TIMEOUT = Duration.minutes(27) const UNLINKED_NOTICE = "This workspace isn't connected to a Maple organization yet — an admin can link it under Integrations in Maple." @@ -461,7 +463,7 @@ const relayMessage = Effect.fn("ChatBotRelay.relayTurn")(function* ( messages: [], recordedAt: now, } - yield* driveChatTurn({ + const followed = yield* driveChatTurn({ // From the cursor the claim answered with, so the turn's own first event is the first one // this sees. events: chatTurnEvents(session, sessionId, claimed.value.cursor), @@ -474,7 +476,11 @@ const relayMessage = Effect.fn("ChatBotRelay.relayTurn")(function* ( chartImageUrl: (ref) => ports.chartImageUrl(orgId, ref), }, onPosted: (messages) => ports.recordTurn({ ...checkpoint, messages }), - }).pipe(Effect.timeout(RELAY_TIMEOUT)) + }).pipe(Effect.timeoutOption(RELAY_TIMEOUT)) + if (Option.isNone(followed)) { + yield* Effect.annotateCurrentSpan({ "maple.chat.relay": "handed_off" }) + yield* ports.handOffTurn(checkpoint) + } }) /** diff --git a/apps/chat-bot/src/socket/ConnectorSocket.test.ts b/apps/chat-bot/src/socket/ConnectorSocket.test.ts new file mode 100644 index 0000000000..29ca9bf4a3 --- /dev/null +++ b/apps/chat-bot/src/socket/ConnectorSocket.test.ts @@ -0,0 +1,208 @@ +/** + * The socket object's lifecycle, against a fake Durable Object state, a fake `WebSocket` and a fake + * connector: what a dial leaves behind, what a tick decides before the socket has opened, and when + * it gives up. + */ +import { socketIngress, type SocketStep } from "@maple/chat-platform" +import { Effect, Schema } from "effect" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" +import type { IngressConnector } from "../config.ts" +import type { ConnectorRelayClient } from "../relay/stub.ts" +import { TEST_SOCKET_ID, TEST_TOKEN_KEY } from "../test-support.ts" +import { ConnectorSocket, OPEN_TIMEOUT_MS } from "./ConnectorSocket.ts" + +const HEARTBEAT_MS = 41_250 + +const FakeState = Schema.Struct({ awaitingAck: Schema.Boolean, heartbeatIntervalMs: Schema.Number }) +type FakeState = typeof FakeState.Type + +/** + * A heartbeat protocol shaped like a real gateway's: a heartbeat sent while the last one is still + * unacknowledged means the connection is a zombie, so the connector asks for a reconnect. + */ +const fakeConnector: IngressConnector = { + id: TEST_SOCKET_ID, + ingress: socketIngress({ + requiredConfig: [{ name: TEST_TOKEN_KEY, secret: true }], + stateSchema: Schema.fromJsonString(FakeState), + initialState: { awaitingAck: false, heartbeatIntervalMs: HEARTBEAT_MS }, + connectUrl: () => "wss://socket.test/gateway", + onOpen: (state: FakeState, now: number): SocketStep => ({ + state: { ...state, awaitingAck: false }, + heartbeatAt: now + state.heartbeatIntervalMs, + }), + onFrame: (state: FakeState, frame: string): SocketStep => ({ + state: frame === "ack" ? { ...state, awaitingAck: false } : state, + }), + onClose: (state: FakeState): SocketStep => ({ + state, + directive: { _tag: "reconnect", closeCode: 4_000 }, + }), + heartbeat: (state: FakeState, now: number): SocketStep => + state.awaitingAck + ? { state, directive: { _tag: "reconnect", closeCode: 4_009 } } + : { + state: { ...state, awaitingAck: true }, + send: ["ping"], + heartbeatAt: now + state.heartbeatIntervalMs, + }, + }), +} + +/** The fields the object reads off a socket event. */ +interface SocketEvent { + readonly data?: string + readonly code?: number + readonly reason?: string +} + +class FakeWebSocket { + static readonly CONNECTING = 0 + static readonly OPEN = 1 + static readonly opened: Array = [] + readyState = FakeWebSocket.CONNECTING + readonly sent: Array = [] + closedWith: number | undefined + private readonly listeners = new Map void>>() + constructor(readonly url: string) { + FakeWebSocket.opened.push(this) + } + addEventListener(type: string, listener: (event: SocketEvent) => void): void { + this.listeners.set(type, [...(this.listeners.get(type) ?? []), listener]) + } + send(frame: string): void { + this.sent.push(frame) + } + close(code: number): void { + this.closedWith = code + if (FakeWebSocket.refuseClose) throw new TypeError("close refused while connecting") + } + static refuseClose = false + open(): void { + this.readyState = FakeWebSocket.OPEN + for (const listener of this.listeners.get("open") ?? []) listener({}) + } +} + +const objectState = (stored: Map = new Map()) => { + const pending: Array> = [] + return { + stored, + pending, + storage: { + get: (key: string) => Promise.resolve(stored.get(key) as T | undefined), + put: (key: string, value: T) => { + stored.set(key, value) + return Promise.resolve() + }, + delete: (key: string) => Promise.resolve(stored.delete(key)), + setAlarm: () => Promise.resolve(), + }, + waitUntil: (promise: Promise) => void pending.push(promise), + blockConcurrencyWhile: (callback: () => Promise) => callback(), + } +} + +const relays: ConnectorRelayClient = { + getByName: () => ({ deliver: () => Effect.void }), +} + +const ENV = { [TEST_TOKEN_KEY]: "token" } +const NOW = 1_800_000_000_000 + +/** What an evicted activation left in storage: a session mid-heartbeat, and its due time. */ +const evicted = () => + new Map([ + ["connectorId", TEST_SOCKET_ID], + ["protocol", JSON.stringify({ awaitingAck: true, heartbeatIntervalMs: HEARTBEAT_MS })], + ["heartbeatAt", NOW - 1_000], + ]) + +beforeEach(() => { + vi.useFakeTimers({ toFake: ["Date"] }) + vi.setSystemTime(NOW) + FakeWebSocket.opened.length = 0 + FakeWebSocket.refuseClose = false + vi.stubGlobal("WebSocket", FakeWebSocket) +}) + +afterEach(() => { + vi.useRealTimers() + vi.unstubAllGlobals() +}) + +describe("ConnectorSocket", () => { + it("does not fire a dead connection's heartbeat on the socket that replaced it", async () => { + const state = objectState(evicted()) + const socket = new ConnectorSocket(state, ENV, relays, [fakeConnector]) + + await socket.ensureConnected(TEST_SOCKET_ID) + await socket.alarm() + await Promise.all(state.pending) + + expect(FakeWebSocket.opened).toHaveLength(1) + // The stale `awaitingAck` would have closed the fresh socket for a missed acknowledgement. + expect(FakeWebSocket.opened[0]?.closedWith).toBeUndefined() + expect(FakeWebSocket.opened[0]?.sent).toEqual([]) + expect(state.stored.has("heartbeatAt")).toBe(false) + }) + + it("sends the first heartbeat once the replacement socket is open and it is due", async () => { + const state = objectState(evicted()) + const socket = new ConnectorSocket(state, ENV, relays, [fakeConnector]) + + await socket.ensureConnected(TEST_SOCKET_ID) + FakeWebSocket.opened[0]?.open() + await Promise.all(state.pending) + vi.setSystemTime(Number(state.stored.get("heartbeatAt"))) + await socket.alarm() + + expect(FakeWebSocket.opened[0]?.sent).toEqual(["ping"]) + expect(FakeWebSocket.opened[0]?.closedWith).toBeUndefined() + }) + + it("arms the handshake timer once the socket opens", async () => { + const state = objectState(evicted()) + const socket = new ConnectorSocket(state, ENV, relays, [fakeConnector]) + + await socket.ensureConnected(TEST_SOCKET_ID) + FakeWebSocket.opened[0]?.open() + await Promise.all(state.pending) + + expect(state.stored.get("heartbeatAt")).toBeGreaterThan(NOW) + expect(FakeWebSocket.opened[0]?.closedWith).toBeUndefined() + }) + + it("gives up on a socket that never opens, and backs off before dialling again", async () => { + const state = objectState(evicted()) + const socket = new ConnectorSocket(state, ENV, relays, [fakeConnector]) + + await socket.ensureConnected(TEST_SOCKET_ID) + await socket.alarm() + expect(FakeWebSocket.opened[0]?.closedWith).toBeUndefined() + + vi.setSystemTime(NOW + OPEN_TIMEOUT_MS) + await socket.alarm() + await Promise.all(state.pending) + + expect(FakeWebSocket.opened[0]?.closedWith).toBe(1_000) + expect(state.stored.get("reconnectAt")).toBeGreaterThan(NOW + OPEN_TIMEOUT_MS) + expect(state.stored.get("attempt")).toBe(1) + }) + + it("drops a socket that refuses the close, and still backs off and dials again", async () => { + const state = objectState(evicted()) + const socket = new ConnectorSocket(state, ENV, relays, [fakeConnector]) + FakeWebSocket.refuseClose = true + + await socket.ensureConnected(TEST_SOCKET_ID) + vi.setSystemTime(NOW + OPEN_TIMEOUT_MS) + await socket.alarm() + await Promise.all(state.pending) + expect(state.stored.get("attempt")).toBe(1) + + vi.setSystemTime(Number(state.stored.get("reconnectAt"))) + await socket.alarm() + expect(FakeWebSocket.opened).toHaveLength(2) + }) +}) diff --git a/apps/chat-bot/src/socket/ConnectorSocket.ts b/apps/chat-bot/src/socket/ConnectorSocket.ts index 492c882397..c9aa122ca0 100644 --- a/apps/chat-bot/src/socket/ConnectorSocket.ts +++ b/apps/chat-bot/src/socket/ConnectorSocket.ts @@ -17,10 +17,9 @@ * * What one step of a connector's protocol CAUSES lives in `./driver.ts`, where a * fake connector drives it without workerd. What is left here is the socket, the - * alarm and the storage keys. The lifecycle this class owns on top of that — - * when to dial, when to back off, how long a fatal directive holds — is not - * covered by a test today; it needs a fake Durable Object state, which is worth - * building the first time one of these rules changes. + * alarm and the storage keys. The lifecycle this class owns on top of that + * (when to dial, how long a dial may take to open, when a heartbeat is due) is + * covered by `./ConnectorSocket.test.ts` against a fake state and socket. * * All socket I/O happens inside this object: the socket is created here, its * handlers run here under the object's own `waitUntil`, and no promise crosses @@ -86,6 +85,12 @@ const WATCHDOG_MS = 60_000 /** Never schedule an alarm closer than this, so a misbehaving step cannot spin. */ const MIN_ALARM_MS = 1_000 +/** + * How long a dialled socket may take to open. A connector's own handshake timer + * starts in `onOpen`, so without this a socket that never opens is never noticed. + */ +export const OPEN_TIMEOUT_MS = 30_000 + /** A connection that stayed up this long counts as established, and resets the backoff. */ const ESTABLISHED_MS = 60_000 @@ -114,6 +119,12 @@ export class ConnectorSocketCallFailed extends Schema.TaggedError()( + "@maple/chat-bot/ConnectorSocketCloseRefused", + { message: Schema.String, cause: Schema.Defect() }, +) {} + const socketCall = (operation: string, call: () => Promise) => Effect.tryPromise({ try: call, @@ -180,6 +191,8 @@ export class ConnectorSocket { */ private generation = 0 private connectedAt: number | undefined + /** When a dialled socket that has not opened yet is given up on. */ + private openBy: number | undefined /** * Steps run one at a time, in arrival order. @@ -204,6 +217,8 @@ export class ConnectorSocket { private readonly ctx: ConnectorSocketState, env: Record, private readonly relays: ConnectorRelayClient, + /** The connectors this object can be; the registry unless a test hands it a fake. */ + private readonly registry: ReadonlyArray = connectors, ) { this.configLayer = workerEnvLayer(env) } @@ -237,6 +252,14 @@ export class ConnectorSocket { if (this.socket === undefined) { const reconnectAt = await this.ctx.storage.get(KEY.reconnectAt) if (reconnectAt === undefined || reconnectAt <= now) await this.connect() + } else if (this.connectedAt === undefined) { + // Dialled and not open: no heartbeat is this connection's yet, only the open deadline. + if (this.openBy !== undefined && this.openBy <= now) { + await this.enqueue(this.generation, (_ingress, state) => ({ + state, + directive: { _tag: "reconnect", closeCode: NORMAL_CLOSE }, + })) + } } else { const heartbeatAt = await this.ctx.storage.get(KEY.heartbeatAt) if (heartbeatAt !== undefined && heartbeatAt <= now) { @@ -272,12 +295,18 @@ export class ConnectorSocket { await this.applyDirective({ _tag: "reconnect", closeCode: NORMAL_CLOSE }, Date.now()) return } + // A heartbeat stored by an evicted activation belongs to the dead connection: due + // before this one's `onOpen` resets the protocol, it would judge the new socket stale. + await this.ctx.storage.delete(KEY.heartbeatAt) const generation = ++this.generation const socket = new WebSocket(url) this.socket = socket this.connectedAt = undefined + this.openBy = Date.now() + OPEN_TIMEOUT_MS socket.addEventListener("open", () => { + if (generation !== this.generation) return this.connectedAt = Date.now() + this.openBy = undefined this.dispatch(generation, (ingress, current, now) => ingress.onOpen(current, now)) }) socket.addEventListener("message", (event) => { @@ -388,9 +417,28 @@ export class ConnectorSocket { const established = this.connectedAt !== undefined && now - this.connectedAt >= ESTABLISHED_MS // Bump first: the close below delivers an event this object no longer wants. this.generation += 1 - this.socket?.close(directive._tag === "reconnect" ? directive.closeCode : NORMAL_CLOSE) + const closing = this.socket this.socket = undefined this.connectedAt = undefined + this.openBy = undefined + // Forgotten before the close: a socket that refuses it (one still connecting) is dropped anyway. + Effect.runSync( + Effect.try({ + try: () => + closing?.close(directive._tag === "reconnect" ? directive.closeCode : NORMAL_CLOSE), + catch: (cause) => + new ConnectorSocketCloseRefused({ message: "The socket refused a close", cause }), + }).pipe( + Effect.catch((error) => + Effect.logWarning(error.message).pipe( + Effect.annotateLogs({ + "error.type": error._tag, + "error.message": String(error.cause), + }), + ), + ), + ), + ) await this.ctx.storage.delete(KEY.heartbeatAt) if (directive._tag === "stop") { @@ -420,6 +468,7 @@ export class ConnectorSocket { now + WATCHDOG_MS, await this.ctx.storage.get(KEY.heartbeatAt), this.socket === undefined ? await this.ctx.storage.get(KEY.reconnectAt) : undefined, + this.openBy, ], (at): at is number => at !== undefined, ) @@ -440,7 +489,7 @@ export class ConnectorSocket { private async resolve(): Promise { const connectorId = await this.ctx.storage.get(KEY.connectorId) if (connectorId === undefined) return undefined - const found = Arr.findFirst(connectors, (candidate) => candidate.id === connectorId) + const found = Arr.findFirst(this.registry, (candidate) => candidate.id === connectorId) if (Option.isNone(found)) return undefined const connector = found.value if (connector.ingress.kind !== "socket") return undefined diff --git a/apps/sandbox/src/executor-scripts.test.ts b/apps/sandbox/src/executor-scripts.test.ts index 9e7bc58c92..2eec70ad2d 100644 --- a/apps/sandbox/src/executor-scripts.test.ts +++ b/apps/sandbox/src/executor-scripts.test.ts @@ -10,6 +10,7 @@ import { lockfilePlanScript, parsePrepareStatus, parseRunOutput, + prepareStatusScript, promoteScript, } from "./executor-scripts" @@ -96,6 +97,37 @@ describe("parsePrepareStatus", () => { }) }) +describe("prepareStatusScript", () => { + /** The script against a scratch root standing in for `/work`. */ + const status = (setup: (state: string) => void) => { + const root = tempDir("maple-executor-status-") + setup(join(root, "prepare")) + return Str.trim(run(Str.replaceAll("/work", root)(prepareStatusScript()), root)) + } + + it("claims a prepare nobody has started", () => { + assert.strictEqual( + status(() => undefined), + "claimed", + ) + }) + + it("reads a fresh claim with no process yet as running", () => { + assert.strictEqual( + status((state) => mkdirSync(state)), + "running", + ) + }) + + it("reads a claim that was never started, past its grace, as lost", () => { + const answer = status((state) => { + mkdirSync(state) + run(`touch -t 200001010000 ${JSON.stringify(state)}`, tmpdir()) + }) + assert.strictEqual(answer, "lost") + }) +}) + describe("parseRunOutput", () => { it("splits the command's output from the trailer, the last trailer winning", () => { const parsed = parseRunOutput( diff --git a/apps/sandbox/src/executor-scripts.ts b/apps/sandbox/src/executor-scripts.ts index 6431ce3e42..5ecfe87989 100644 --- a/apps/sandbox/src/executor-scripts.ts +++ b/apps/sandbox/src/executor-scripts.ts @@ -212,7 +212,8 @@ export const prepareRunner = (sha: string): string => { /** * Where prepare stands: `ready `, `running`, `failed ` (then the log's end), `lost` - * (it stopped without an exit code) or `claimed` (nothing ran yet; the caller starts it). + * (it stopped without an exit code, or was claimed and never started) or `claimed` (nothing ran + * yet; the caller starts it). The caller resets a failed or lost prepare, so the next call claims. */ export const prepareStatusScript = (): string => { const s = PREPARE_STATE_DIR @@ -228,6 +229,11 @@ export const prepareStatusScript = (): string => { ` if kill -0 "$(cat ${s}/pid)" 2>/dev/null; then echo running; else echo lost; fi`, " exit 0", "fi", + // Claimed but never started: the Durable Object went away in between. Give it two minutes. + `if [ -d ${s} ]; then`, + ` if [ -z "$(find ${s} -maxdepth 0 -mmin +2)" ]; then echo running; else echo lost; fi`, + " exit 0", + "fi", `if mkdir -p /work && mkdir ${s} 2>/dev/null; then echo claimed; else echo running; fi`, ], "\n", diff --git a/apps/sandbox/src/executor-session.test.ts b/apps/sandbox/src/executor-session.test.ts index 4f6727ccae..33e6f5492f 100644 --- a/apps/sandbox/src/executor-session.test.ts +++ b/apps/sandbox/src/executor-session.test.ts @@ -118,7 +118,10 @@ describe("prepare", () => { it.effect("reports a failed prepare with its log, and never the credential", () => Effect.gen(function* () { - const fake = fakeHost([[isStatus, exited(`failed 11\nfatal: auth failed for ${TOKEN}\n`)]]) + const fake = fakeHost([ + [isStatus, exited(`failed 11\nfatal: auth failed for ${TOKEN}\n`)], + [(script) => Str.startsWith("rm -rf")(script), exited("")], + ]) const answer = yield* prepare(fake.host, prepareRequest) assert.strictEqual(answer._tag, "ExecutorPrepareFailed") if (answer._tag !== "ExecutorPrepareFailed") return @@ -127,6 +130,15 @@ describe("prepare", () => { }), ) + it.effect("forgets a failed prepare as it reports it, so the next call prepares again", () => + Effect.gen(function* () { + const fake = fakeHost([[isStatus, exited("failed 11\n")]]) + // The reset has no answer here: it failing must not hide the failure being reported. + assert.strictEqual((yield* prepare(fake.host, prepareRequest))._tag, "ExecutorPrepareFailed") + assert.isTrue(Arr.some(fake.execs, (script) => Str.startsWith("rm -rf")(script))) + }), + ) + it.effect("clears a prepare that stopped without finishing so the next call starts again", () => Effect.gen(function* () { const fake = fakeHost([ diff --git a/apps/sandbox/src/executor-session.ts b/apps/sandbox/src/executor-session.ts index 91e419769f..0ff7c58670 100644 --- a/apps/sandbox/src/executor-session.ts +++ b/apps/sandbox/src/executor-session.ts @@ -98,6 +98,9 @@ export const prepare = ( ), }) case "failed": + // Forgotten as it is reported, like a failed clone: the next call prepares again. A reset + // that fails leaves the failure to be reported again, which is still the right answer. + yield* host.exec(resetPrepareScript(), STATUS_TIMEOUT_MS).pipe(Effect.ignore) return new ExecutorPrepareFailed({ message: redact( `preparing ${checkout.repository} failed (exit ${status.code}): ${status.log}`, diff --git a/packages/chat-platform/src/driver.test.ts b/packages/chat-platform/src/driver.test.ts index aaed22d143..22b0c75574 100644 --- a/packages/chat-platform/src/driver.test.ts +++ b/packages/chat-platform/src/driver.test.ts @@ -108,6 +108,51 @@ const working = (blocks: ReadonlyArray): ReadonlyArray => ) describe("driveChatTurn", () => { + it.effect("leaves the streaming render on interrupt, with the ending on the last posted message", () => + Effect.gen(function* () { + const chat = recorder(40) + const fiber = yield* Effect.forkChild( + driveChatTurn({ + events: Stream.concat( + timeline([ + [NOW, event(1, { type: "turn-start", messageId: "a1" })], + [NOW, event(2, { type: "text-delta", messageId: "a1", text: "First part." })], + [ + Duration.millis(1500), + event(3, { + type: "text-delta", + messageId: "a1", + text: " A much longer second part that needs another message.", + }), + ], + ]), + Stream.never, + ), + messageId: "a1", + outbound: chat.outbound, + target, + context, + }), + ) + yield* TestClock.adjust("1 second") + const posts = Arr.filter(chat.calls, (call) => call.verb === "post").length + yield* TestClock.adjust("600 millis") + yield* Fiber.interrupt(fiber) + + // No message is posted on the way out: the turn may not grow once nobody follows it. + expect(Arr.filter(chat.calls, (call) => call.verb === "post").length).toBe(posts) + const last = Arr.getUnsafe(chat.calls, chat.calls.length - 1) + expect(last.phase).toBe("final") + expect(Arr.filter(last.blocks, (block) => block.kind === "notice")).toEqual([ + { + kind: "notice", + tone: "info", + text: "Still working on this. The full answer will replace this message when it is ready.", + }, + ]) + }), + ) + it.effect("posts a placeholder, coalesces deltas into throttled edits, and flushes at the end", () => Effect.gen(function* () { const chat = recorder() diff --git a/packages/chat-platform/src/driver.ts b/packages/chat-platform/src/driver.ts index d7187382d7..e992e77b89 100644 --- a/packages/chat-platform/src/driver.ts +++ b/packages/chat-platform/src/driver.ts @@ -86,53 +86,74 @@ export const driveChatTurn = Effect.fn("ChatPlatform.driveChatTurn")(function* < running: true, }) - const flush = gate.withPermit( - Effect.gen(function* () { - // Messages an earlier run posted already say more than a render of no events would. - if (resumed && transcript.seq === 0) return - const { ending, running } = yield* Ref.updateAndGet(turn, (state) => ({ ...state, dirty: false })) - const blocks: Array = Option.match( - Arr.findFirst(transcript.messages, (candidate) => candidate.id === messageId), - { onNone: () => [], onSome: (message) => [...renderChatMessage(message, context, running)] }, - ) - if (ending !== null) blocks.push(ending) - // A finished turn must never be left under the placeholder: a model that stops after a - // tool call without a closing word renders as nothing at all, and "Working on it…" would - // then be the last thing the channel ever says about it. - if (blocks.length === 0) blocks.push(running ? PENDING : NOTHING_SAID) + /** `mayPost` false only edits what is posted: an interrupted driver must not grow the turn. */ + const render = (mayPost: boolean) => + gate.withPermit( + Effect.gen(function* () { + // Messages an earlier run posted already say more than a render of no events would. + if (resumed && transcript.seq === 0) return + const { ending, running } = yield* Ref.updateAndGet(turn, (state) => ({ + ...state, + dirty: false, + })) + const blocks: Array = Option.match( + Arr.findFirst(transcript.messages, (candidate) => candidate.id === messageId), + { + onNone: () => [], + onSome: (message) => [...renderChatMessage(message, context, running)], + }, + ) + if (ending !== null) blocks.push(ending) + // A finished turn must never be left under the placeholder: a model that stops after a + // tool call without a closing word renders as nothing at all, and "Working on it…" would + // then be the last thing the channel ever says about it. + if (blocks.length === 0) blocks.push(running ? PENDING : NOTHING_SAID) - const groups = splitBlocks(blocks, outbound.limits.maxMessageChars) - const phase: ChatWritePhase = running ? "streaming" : "final" - // Messages a flush posts extend `posted` as it goes; the count is fixed up front. - yield* Effect.forEach( - Array.from({ length: Math.max(groups.length, posted.length) }, (_, index) => index), - (index) => - Effect.gen(function* () { - // A retraction can shrink a turn below a message it had already needed, so a surplus - // message is emptied rather than left holding text the turn no longer says. - const group = groups[index] ?? NO_BLOCKS - const rendered = renderedKey(group) - const existing = posted[index] - if (existing === undefined) { - const postedMessage = yield* transport.post(target, group, phase) - posted.push(postedMessage) - sent.push(rendered) - // At least once: a run lost between the post and this report posts it again later. - if (options.onPosted !== undefined) yield* options.onPosted([...posted]) - return - } - // Only what changed. A turn cut into three messages would otherwise spend three edits - // per tick, almost all of them rewriting a message with what it already says — and a - // platform's edit budget is per channel, not per message. A live message is the - // exception at the end: it shows as still being written until its final write. - if (sent[index] === rendered && (running || outbound.liveMessages !== true)) return - yield* transport.edit(existing, group, phase) - sent[index] = rendered - }), - { discard: true }, - ) - }), - ) + const split = splitBlocks(blocks, outbound.limits.maxMessageChars) + // Kept to the posted messages, an interrupted render puts its ending on the last of them. + const cut = !mayPost && ending !== null && posted.length > 0 && split.length > posted.length + const groups = cut + ? [ + ...split.slice(0, posted.length - 1), + [...(split[posted.length - 1] ?? NO_BLOCKS), ending], + ] + : split + const phase: ChatWritePhase = running ? "streaming" : "final" + // Messages a flush posts extend `posted` as it goes; the count is fixed up front. + yield* Effect.forEach( + Array.from( + { length: mayPost ? Math.max(groups.length, posted.length) : posted.length }, + (_, index) => index, + ), + (index) => + Effect.gen(function* () { + // A retraction can shrink a turn below a message it had already needed, so a surplus + // message is emptied rather than left holding text the turn no longer says. + const group = groups[index] ?? NO_BLOCKS + const rendered = renderedKey(group) + const existing = posted[index] + if (existing === undefined) { + const postedMessage = yield* transport.post(target, group, phase) + posted.push(postedMessage) + sent.push(rendered) + // At least once: a run lost between the post and this report posts it again later. + if (options.onPosted !== undefined) yield* options.onPosted([...posted]) + return + } + // Only what changed. A turn cut into three messages would otherwise spend three edits + // per tick, almost all of them rewriting a message with what it already says, and a + // platform's edit budget is per channel, not per message. A live message is the + // exception at the end: it shows as still being written until its final write. + if (sent[index] === rendered && (running || outbound.liveMessages !== true)) + return + yield* transport.edit(existing, group, phase) + sent[index] = rendered + }), + { discard: true }, + ) + }), + ) + const flush = render(true) const onEvent = Effect.fnUntraced(function* (event: ChatEvent) { const now = yield* Clock.currentTimeMillis @@ -175,28 +196,40 @@ export const driveChatTurn = Effect.fn("ChatPlatform.driveChatTurn")(function* < // notice before the stream completes. An earlier turn's does not stop the stream. const endsThisTurn = (event: ChatEvent) => event.type === "turn-end" && event.task === undefined && event.messageId === messageId + // Whatever ended the turn, the last render is the finished message: the answer, and none of the + // working. Set here rather than on `turn-end`, so a stream that died, and never reaches one, + // settles the same way. Cleared only once the throttle fiber is gone, so no flush reads it half + // way through a turn it has not finished. + const finish = (mayPost: boolean) => + Effect.gen(function* () { + yield* Fiber.interrupt(throttled) + yield* Ref.update(turn, (state) => ({ ...state, running: false })) + yield* render(mayPost) + yield* Effect.annotateCurrentSpan({ + "maple.chat.turn.end_reason": (yield* Ref.get(turn)).endReason, + "maple.chat.messages": posted.length, + }) + }) + const endWith = (ending: ChatNoticeBlock, endReason: string) => + Ref.update(turn, (state) => (state.ending === null ? { ...state, ending, endReason } : state)) const outcome = yield* Stream.runForEach( options.events.pipe(Stream.takeUntil(endsThisTurn)), onEvent, ).pipe( // A stream that dies has no `turn-end` to close on, and the reader would be left looking at // the placeholder forever. Say so, flush, and then fail. - Effect.tapError(() => - Ref.update(turn, (state) => (state.ending === null ? { ...state, ending: DISCONNECTED } : state)), + Effect.tapError(() => endWith(DISCONNECTED, "disconnected")), + // Interrupted (a caller's timeout): the message still leaves its streaming render, saying the + // answer is not in yet. Best effort, since nothing is left to report a failure to. + Effect.onInterrupt(() => + endWith(STOPPED_FOLLOWING, "interrupted").pipe( + Effect.andThen(finish(false)), + Effect.catchCause(report("A turn's final message could not be written")), + ), ), Effect.exit, ) - yield* Fiber.interrupt(throttled) - // Whatever ended the turn, the last render is the finished message: the answer, and none of the - // working. Set here rather than on `turn-end`, so a stream that died — and never reaches one — - // settles the same way. Cleared only once the throttle fiber is gone, so no flush reads it half - // way through a turn it has not finished. - yield* Ref.update(turn, (state) => ({ ...state, running: false })) - yield* flush - yield* Effect.annotateCurrentSpan({ - "maple.chat.turn.end_reason": (yield* Ref.get(turn)).endReason, - "maple.chat.messages": posted.length, - }) + yield* finish(true) return yield* outcome }) @@ -241,6 +274,13 @@ const NOTHING_SAID: ChatNoticeBlock = { text: "Finished without a reply.", } +/** The render a turn is left with when its driver is stopped before the turn ends. */ +const STOPPED_FOLLOWING: ChatNoticeBlock = { + kind: "notice", + tone: "info", + text: "Still working on this. The full answer will replace this message when it is ready.", +} + const DISCONNECTED: ChatNoticeBlock = { kind: "notice", tone: "error",