diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index ab6a13b1c..80074c428 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -18,6 +18,166 @@ entry. See `CONTRIBUTING.md` § Releases & changelog. ## [Unreleased] +### Fixed — streamed turns no longer report a bare error after a tool already committed (#506) + +- Root-cause fix for issue #506's actual one-click repro (the earlier + reconciliation work below only helped on a *retry*). `chatStreamInner` + in `middleware/packages/harness-orchestrator/src/orchestrator.ts` wraps + its whole per-turn iteration loop — tool dispatch and every subsequent + `streamMessageEvents` call — in a single `try`/`catch`. Any exception + caught there unconditionally yielded a bare `{ type: 'error' }` event, + even when it happened in a LATER iteration (e.g. the model call that + generates the natural-language confirmation), after an EARLIER + iteration's tool call had already committed its side effect and already + yielded a successful `tool_result`. A user who clicked a create action + exactly once would have it created server-side and still see a generic + "Etwas ist schief gegangen" with zero diagnostic value — the false + negative the issue was filed against. The streaming iteration loop now + tracks, generically and tool-agnostically (by name only, no per-tool + special-casing), whether at least one `tool_result` succeeded + (`isError` falsy) this turn. When the catch block is reached with at + least one such committed result recorded, it now yields a `done` event + instead — `ChatStreamEvent`'s existing normal-completion shape, + already rendered correctly by every channel adapter — with an honest + answer naming the tool(s) that completed and stating that the turn + itself could not finish generating a follow-up response. It does not + claim the whole turn succeeded, and it does not fabricate tool-specific + detail it doesn't generically have. The underlying error is still + `console.error`-logged exactly as before for server-side diagnostics; + only the event yielded to the caller changes. A turn where nothing + committed yet (the genuine-failure case — e.g. the very first model + call fails, or the tool call itself errored) still yields `{ type: + 'error' }` unchanged. Together with the reconciliation fix below, this + closes #506 for both the one-click repro and the retry-duplication + case; the correlation-id/error-token secondary ask remains explicitly + out of scope (see below). +- Review follow-up: the emergency `done` yielded from the catch block above + did not call `this.sessionLogger.log(...)` first — the ONE thing every + other `done`-emission site in `chatStreamInner` does before yielding (see + `SessionLogger`'s doc comment: the transcript is what lets a follow-up + turn recall prior discussion, and what survives a mid-turn crash). For a + tool whose side effect isn't idempotently reconciled the way routine-create + now is (e.g. `send_email`, `book_meeting`), an unlogged commit meant the + *next* turn had no record it happened and could re-invoke the same tool — + the exact duplicate-side-effect class of bug this fix exists to prevent, + reintroduced by the fix's own new code path. The emergency-`done` path now + calls `sessionLogger.log(...)` with the same argument shape as the other + sites (`scope`, `userMessage`, `assistantAnswer`, `toolCalls`, + `iterations`, `entityRefs`, optional `userId`/`runTrace`), best-effort + (a logging failure is caught and logged, never swallows the `done`), and + surfaces `turnId`/`runTrace` on the yielded event when persistence + succeeded. `committedToolReporting.test.ts` now constructs the test + orchestrator WITH a recording `sessionLogger` (the prior 2 tests built one + without any logger at all, which is why the gap was invisible) and asserts + the log call happened, with matching `scope`/`userMessage`/ + `assistantAnswer`/`toolCalls`/`iterations`, plus that a genuine failure + (nothing committed) still does not log. +- Review follow-up: the fix above tracks `committedToolNames` generically — + ANY successful `tool_result` this turn counts as "committed," with no + distinction between a read-only tool and a mutating one. A reviewer raised + the concrete scenario where a read-only tool (e.g. `list_routines`) + succeeds and a LATER, more consequential tool call then never runs because + of a transient failure in the model call that would have requested it — + the turn still reports `done`. This tradeoff — generic-across-all-tools + vs. narrowed-to-routine-create-only vs. dropping the orchestrator fix + entirely — was weighed and resolved in favor of keeping the current + generic, tool-agnostic behavior across all tools, accepting the residual + risk described above in exchange for fixing the false-negative-on-success + bug for every side-effecting tool, not just routine creation. This is now + documented as a deliberate decision (not an oversight) directly in the + code, on both `committedToolNames`'s + declaration and the catch block's done-vs-error branch in + `orchestrator.ts`, and pinned by a new `committedToolReporting.test.ts` + case (`reports done even when a later intended action never ran (accepted + tradeoff, see code comment)`) that exercises exactly this multi-tool + scenario. No production logic changed in this round. + +### Fixed — routine create no longer reports failure for a retry that already succeeded (#506) + +- `RoutineRunner.createRoutine` previously let a retried `create` (e.g. after + the turn's own confirmation never made it back over the channel) fall + through to `RoutineNameConflictError` — a message with no diagnostic value + that nudged the caller toward trying again under a different name and + actually duplicating the routine. It now reconciles: on a name conflict it + looks up the existing row (`RoutineStore.getByName`, new) and, if the + `cron`/`prompt`/`channel`/`timeoutMs` match what was just requested, + returns that row instead of raising — the earlier call already succeeded, + so the retry now sees success too. Reconciliation only fires against an + `active` existing row: a paused/inactive same-name row with otherwise + identical fields still raises `RoutineNameConflictError`, because that is + a genuine, separate collision (e.g. a paused "demo" routine plus a new, + deliberate create under the same name), not the caller's own in-flight + retry — silently reconciling there would report a successful create with + no active schedule, which is a worse instance of the exact + false-negative/false-positive problem this issue was filed to fix. + Reconciliation deliberately does not additionally gate on the existing + row's age/`createdAt`; see the code comment in `createRoutine` for why. A + conflict with genuinely different fields still raises + `RoutineNameConflictError` as before. Threading a + request/trace correlation id through routine-turn error responses + end-to-end (the issue's secondary ask) remains open — it would require a + new field on the shared `ChatTurnInput`/`ChatTurnResult` contract + (`@omadia/channel-sdk`) plus support in every channel adapter, which is + broader than this fix. The literal error wording shown in Teams + ("Etwas ist schief gegangen …") lives in the external Teams-channel + adapter plugin and is out of scope for this repo. + `isSameRoutineRequest`'s field comparison omitted `outputTemplate` — an + independently-settable object field on both `Routine` and + `CreateRoutineInput` (Phase C structured-output templates). A retried + create that agreed on `cron`/`prompt`/`channel`/`timeoutMs` but carried a + *different* `outputTemplate` (e.g. the caller adding or changing the + structured template on an existing schedule) would reconcile to the old + row and silently discard the new template while reporting success — the + exact class of bug this issue exists to eliminate, on a field the fix's + own comparison had missed. `isSameRoutineRequest` now compares + `outputTemplate` too, via `node:util`'s `isDeepStrictEqual` (it is an + object, so reference/`===` equality is not sufficient); an identical + template (including the `null`/`null` case) still reconciles as before. + The reconciliation check also ran too late: `createRoutine` evaluated the + per-user quota (`countActiveForUser`) *before* attempting `store.create()`, + so a retry from a user already sitting at `maxActivePerUser` — exactly the + state their own successful-but-unconfirmed first call left them in — was + rejected with `RoutineQuotaExceededError` before it ever reached the + conflict-reconciliation logic, resurfacing the same false-negative under a + different exception type. `createRoutine` now looks up + `RoutineStore.getByName` and reconciles a same-request, `active` retry + *before* the quota check and before calling `store.create()` at all — no + new row is needed for a retry that already succeeded. The quota check + still applies to every genuinely new routine request. The reconciliation + logic in the `store.create()` catch block is unchanged and remains the + necessary race-safety net for a concurrent request that creates the + matching row between this proactive lookup and the insert. + `isSameRoutineRequest` also excluded `conversationRef` from its + comparison, reasoning it was a delivery-mechanism detail the caller + doesn't control byte-for-byte. That's wrong on the cold-start outreach + path: `ManageRoutineTool.handleCreate` resolves `conversationRef` from + a caller-supplied `targetEmail` via `buildEmailColdStartTarget` before + calling `createRoutine`, so it *is* caller-specified there. A create for + a new `targetEmail` that otherwise matched an existing active routine + (same tenant/user/name/cron/prompt/channel/timeoutMs/`outputTemplate`) + would silently reconcile to the existing row and report success, while + the new recipient was never set up and the routine kept messaging the + original one — a silent-wrong-recipient bug. `isSameRoutineRequest` now + compares `conversationRef` too, via `isDeepStrictEqual` (same rationale + as `outputTemplate`: it is an object, and `buildEmailColdStartTarget` + resolves deterministically per email, so deep equality correctly + distinguishes a true retry from a different-recipient request). +- Review follow-up: `RoutineStore.create()` normalizes an omitted + `conversationRef` to `{}` before persisting it (and reads it back the + same way — `JSON.stringify(input.conversationRef ?? {})`), but + `isSameRoutineRequest`'s new `conversationRef` comparison above compared + the stored (normalized) value against the RAW retry input with no + equivalent `?? {}` default, unlike `timeoutMs` and `outputTemplate`, + which already apply the same default the store itself uses. On the + ordinary (non-cold-start) create path — where `conversationRef` is + legitimately `undefined`/omitted both on the original call and the retry, + since only the `targetEmail` cold-start branch sets a non-default value — + the stored `{}` never matched the retry's raw `undefined`, so the retry + fell through to `RoutineNameConflictError`, reintroducing the exact + false-negative issue #506 exists to fix for that path. + `isSameRoutineRequest` now applies the same `?? {}` normalization the + store uses: `isDeepStrictEqual(existing.conversationRef, input.conversationRef ?? {})`. + ### Fixed — Teams-uploaded images now reach the model as vision input (#504, #505) - Teams delivers inbound images via a Tigris `storage_key` + `[attachments-info]` @@ -149,7 +309,6 @@ entry. See `CONTRIBUTING.md` § Releases & changelog. - Purely additive — `computeHealthScore` (the diff-based drift score `driftWorker.ts` persists) is untouched. Builder UI wiring and `driftWorker.ts` snapshot wiring are deferred to follow-up work; see #499. - ### Fixed — templates v2 review round 3: owner-aware publish vs. auth timing (#478) - The save-as-template dialog no longer reads the viewer's own template id as diff --git a/middleware/packages/harness-orchestrator/src/orchestrator.ts b/middleware/packages/harness-orchestrator/src/orchestrator.ts index 3d2947d35..a95cba1ca 100644 --- a/middleware/packages/harness-orchestrator/src/orchestrator.ts +++ b/middleware/packages/harness-orchestrator/src/orchestrator.ts @@ -3678,6 +3678,32 @@ export class Orchestrator { const textParts: string[] = []; // One forced file-build retry per turn (see fileAnnouncedButNotBuilt). let fileForceRetried = false; + // Issue #506 — generic, tool-agnostic record of which tool(s) already + // committed a successful side effect THIS turn (a `tool_result` yielded + // with `isError` falsy). If a LATER exception (a subsequent model call, + // the nudge pipeline, ...) lands in the catch below, this lets it report + // an honest `done` instead of discarding a real, already-committed + // action behind a bare `error`. Never special-cases a tool by name. + // + // Deliberate tradeoff (issue #506) — not an oversight: this list is + // populated by ANY successful `tool_result`, with no distinction + // between a read-only tool (e.g. `list_routines`) and a mutating one + // (e.g. `manage_routine` create/update). Concrete residual risk: a + // read-only tool succeeds early in the turn, then a LATER, more + // consequential tool call never runs because a transient failure hits + // the model call that would have requested it — the turn is still + // reported `done` (see the catch block below), even though the user's + // actual intended mutating action may never have happened. + // Two narrower alternatives were considered and rejected: (a) scoping + // this tracking to routine-create only sidesteps the residual risk but + // leaves the same false-negative bug unfixed for every other mutating + // tool (send_email, book_meeting, ...); (b) dropping this fix and + // always reporting `error` here regresses to issue #506's original, + // reported symptom for every tool. Kept generic and tool-agnostic + // across all tools as the better tradeoff; re-evaluate before + // narrowing it. + const committedToolNames: string[] = []; + let lastIterationIndex = 0; const traceCollector = input.sessionScope ? new RunTraceCollector({ @@ -3739,6 +3765,7 @@ export class Orchestrator { } try { for (let iteration = 0; iteration < this.maxIterations; iteration++) { + lastIterationIndex = iteration; yield { type: 'iteration_start', iteration }; // Mirror BuilderAgent: the per-iteration boundary is also when the // observer's iteration counter resets, so its consumers (heartbeat @@ -4089,6 +4116,14 @@ export class Orchestrator { s.isError = winner.output.startsWith('Error:'); s.durationMs = Date.now() - s.started; this.finishSlotInvocation(s, traceCollector); + // Issue #506 — record the committed side effect before yielding, + // generically (name only, no tool-specific payload inspection). + if (!s.isError) { + const name = s.use.name; + if (typeof name === 'string' && !committedToolNames.includes(name)) { + committedToolNames.push(name); + } + } yield { type: 'tool_result', id: s.use.id, @@ -4264,10 +4299,89 @@ export class Orchestrator { '[orchestrator] turn failed:', err instanceof Error ? (err.stack ?? err.message) : err, ); - yield { - type: 'error', - message: err instanceof Error ? err.message : String(err), - }; + // Issue #506 — a tool call earlier in this turn may have already + // committed a real side effect (e.g. created a record) even though a + // LATER step of the SAME turn (a subsequent model call, the nudge + // pipeline, ...) then threw. Reporting a bare `error` in that case is + // a false negative: the action succeeded, only the turn's own + // bookkeeping failed afterwards. Report `done` instead — honest that + // the action(s) completed but the turn itself didn't finish cleanly. + // Generic across every tool; no tool-specific detail is fabricated. + // A genuine failure (nothing committed yet) still yields `error`, + // unchanged from today. + // + // Deliberate tradeoff — not an oversight: this done-vs-error branch + // trusts ANY entry in `committedToolNames` equally, read-only or + // mutating (see the fuller tradeoff comment on `committedToolNames`'s + // declaration above). Accepted residual risk: a benign read-only + // success earlier in the turn can mask a later, more consequential + // mutation that was silently skipped, and this branch will still + // report `done`. Kept generic across all tools rather than narrowed + // to routine-create only or dropped entirely, because reverting to + // always-`error` here would leave issue #506's reported + // false-negative-on-success bug unfixed for every side-effecting + // tool, not just routine creation. + if (committedToolNames.length > 0) { + const toolList = committedToolNames.join(', '); + const answer = + committedToolNames.length === 1 + ? `The requested action (${toolList}) completed successfully, but the turn could not finish generating a follow-up response.` + : `The requested actions (${toolList}) completed successfully, but the turn could not finish generating a follow-up response.`; + const iterations = lastIterationIndex + 1; + // Issue #506 (review follow-up) — every OTHER `done`-emission site + // in this function persists the exchange via `sessionLogger.log()` + // BEFORE yielding (see the success path above, the choice-card + // path, and direct-line). This emergency path is specifically for + // the case where a tool already committed a real side effect, so + // skipping the log here would be the one `done` path that leaves + // that commitment unrecorded — the next turn's model would have no + // memory of it and could re-invoke the same tool, reintroducing the + // duplicate-side-effect bug issue #506 exists to prevent. Same + // call shape as the other sites; best-effort like all of them. + const restoredAnswer = await restorePromptForPersistence( + privacyForPrompt, + answer, + ); + const runTrace = traceCollector?.finish({ + iterations, + status: 'success', + }); + let persistedTurnId: string | undefined; + if (this.sessionLogger && input.sessionScope) { + const entityRefs = entityCollection?.drain() ?? []; + try { + const logged = await this.sessionLogger.log({ + scope: input.sessionScope, + userMessage: input.userMessage, + assistantAnswer: restoredAnswer, + toolCalls, + iterations, + entityRefs, + ...(input.userId ? { userId: input.userId } : {}), + ...(runTrace ? { runTrace } : {}), + }); + persistedTurnId = logged.turnExternalId; + } catch (logErr) { + console.error( + '[orchestrator] session log failed (continuing with emergency done):', + logErr instanceof Error ? logErr.message : logErr, + ); + } + } + yield { + type: 'done', + answer: restoredAnswer, + toolCalls, + iterations, + ...(persistedTurnId ? { turnId: persistedTurnId } : {}), + ...(runTrace ? { runTrace } : {}), + }; + } else { + yield { + type: 'error', + message: err instanceof Error ? err.message : String(err), + }; + } } finally { entityCollection?.drain(); } diff --git a/middleware/src/plugins/routines/routineRunner.ts b/middleware/src/plugins/routines/routineRunner.ts index a78d0636e..f640a6221 100644 --- a/middleware/src/plugins/routines/routineRunner.ts +++ b/middleware/src/plugins/routines/routineRunner.ts @@ -1,3 +1,5 @@ +import { isDeepStrictEqual } from 'node:util'; + import type { ChatTurnInput, ChatTurnResult, @@ -46,11 +48,12 @@ import type { RoutineRunsStore, RoutineRunTrigger, } from './routineRunsStore.js'; -import type { - CreateRoutineInput, - Routine, - RoutineRunStatus, - RoutineStore, +import { + RoutineNameConflictError, + type CreateRoutineInput, + type Routine, + type RoutineRunStatus, + type RoutineStore, } from './routineStore.js'; /** @@ -299,6 +302,34 @@ export class RoutineRunner { if (!this.senders.get(input.channel)) { throw new UnknownChannelError(input.channel); } + + // Issue #506: check for a reconcile-eligible retry BEFORE the quota + // gate. A retried `createRoutine` call for a routine that already + // exists and is already active doesn't need a fresh slot — it needs + // nothing at all, because the work is already done. Without this + // proactive check, a user sitting exactly at `maxActivePerUser` whose + // confirmation got dropped (the scenario issue #506 targets) would + // have their retry rejected by the quota check below before it ever + // reaches the `RoutineNameConflictError` reconciliation in the + // `store.create()` catch block, surfacing a different but equally + // wrong "you're at capacity" error instead of the routine they already + // created. The `catch` block's reconciliation stays as-is: it remains + // the necessary race-safety net for a concurrent request that creates + // the matching row between this lookup and the `store.create()` call + // below. + const existing = await this.store.getByName( + input.tenant, + input.userId, + input.name, + ); + if ( + existing && + existing.status === 'active' && + isSameRoutineRequest(existing, input) + ) { + return existing; + } + const active = await this.store.countActiveForUser( input.tenant, input.userId, @@ -307,7 +338,56 @@ export class RoutineRunner { throw new RoutineQuotaExceededError(this.maxActivePerUser); } - const row = await this.store.create(input); + let row: Routine; + try { + row = await this.store.create(input); + } catch (err) { + // Issue #506: a retried `create` for the same (tenant, user, name) — + // e.g. the model or user retrying after the turn's own confirmation + // never made it back over the channel — hits the unique-name + // constraint here rather than actually duplicating anything. Rather + // than reporting that retry as a failure (which the caller can only + // read as "try again", pushing toward a real duplicate under a + // different name), reconcile: if the existing row is the same + // request, treat this call as already-succeeded and return it. A + // conflicting row with different cron/prompt/channel is a genuine + // name collision and still surfaces as an error. We deliberately do + // NOT additionally gate on `existing.createdAt` recency: the + // `status === 'active'` check below plus the field-by-field + // comparison in `isSameRoutineRequest` already narrow this to "an + // active routine with byte-for-byte identical cron/prompt/channel/ + // timeout" — a false-positive reconciliation on an old-but-still- + // active routine is the caller re-issuing an identical create, which + // is the same "already succeeded" case regardless of age. Adding a + // time window would only reintroduce a class of false negatives + // (rejecting a legitimate late retry) without closing any real gap. + if (err instanceof RoutineNameConflictError) { + const existing = await this.store.getByName( + input.tenant, + input.userId, + input.name, + ); + // Only reconcile against an `active` row. A paused/inactive row + // with matching fields is not "my own in-flight retry" — it's a + // genuine, separate collision (e.g. the user paused an earlier + // "demo" routine and is now deliberately creating a new one under + // the same name). Reconciling there would silently hand back a + // routine that has no active schedule, which is a worse version of + // the exact false-negative/false-positive problem issue #506 was + // filed to fix. Only an `active` existing row can plausibly be + // "the create that already succeeded", so only that case skips the + // error. + if ( + existing && + existing.status === 'active' && + isSameRoutineRequest(existing, input) + ) { + return existing; + } + } + throw err; + } + try { this.registerInScheduler(row); } catch (err) { @@ -715,6 +795,44 @@ function emptySlotsFor( return slots; } +/** + * Whether `existing` (the row that already holds the unique-name slot) is + * indistinguishable, from the caller's point of view, from what `input` + * asked to create. Compared on the fields the user/model actually + * specified — `cron`, `prompt`, `channel`, the resolved `timeoutMs` + * (matching `store.create`'s own default so an omitted vs. explicit + * default value doesn't defeat the comparison), `outputTemplate` + * (Phase C structured-output template — an independently-settable object, + * so it needs a structural/deep comparison rather than `===`; two calls + * that agree on everything else but differ on `outputTemplate` are the + * caller asking to change the template on an existing schedule, not a + * retry, and must still surface as a name conflict), and `conversationRef`. + * `conversationRef` is also caller-specified on the cold-start outreach + * path: `ManageRoutineTool.handleCreate` derives it from `targetEmail` via + * `buildEmailColdStartTarget`, which resolves deterministically per email — + * same `targetEmail` on both calls produces the same `conversationRef` + * structure (a true retry), while a different `targetEmail` produces a + * different one (a genuine new request, e.g. the same routine name aimed + * at a different recipient). Excluding it would let a create for a new + * recipient silently reconcile to — and return as "created" — an existing + * row still routed to the *original* recipient, a silent-wrong-recipient + * bug. Deep comparison (not `===`) for the same reason as `outputTemplate`: + * it is `unknown`/an object, not a primitive. + */ +function isSameRoutineRequest( + existing: Routine, + input: CreateRoutineInput, +): boolean { + return ( + existing.cron === input.cron && + existing.prompt === input.prompt && + existing.channel === input.channel && + existing.timeoutMs === (input.timeoutMs ?? 600_000) && + isDeepStrictEqual(existing.outputTemplate, input.outputTemplate ?? null) && + isDeepStrictEqual(existing.conversationRef, input.conversationRef ?? {}) + ); +} + function errMsg(err: unknown): string { if (err instanceof Error) return err.message; return String(err); diff --git a/middleware/src/plugins/routines/routineStore.ts b/middleware/src/plugins/routines/routineStore.ts index 23dc242d8..336e853a4 100644 --- a/middleware/src/plugins/routines/routineStore.ts +++ b/middleware/src/plugins/routines/routineStore.ts @@ -214,6 +214,28 @@ export class RoutineStore { return row ? rowToRoutine(row) : null; } + /** + * Look up a single routine by its (tenant, user_id, name) unique key — + * the same triple `create()` enforces via `routines_user_name_unique`. + * Used by the runner to reconcile a `RoutineNameConflictError` against + * the row that already won the race (see issue #506: a retried create + * after an ambiguous confirmation must not be told it failed). + */ + async getByName( + tenant: string, + userId: string, + name: string, + ): Promise { + const result = await this.pool.query( + `SELECT ${SELECT_COLUMNS} + FROM routines + WHERE tenant = $1 AND user_id = $2 AND name = $3`, + [tenant, userId, name], + ); + const row = result.rows[0]; + return row ? rowToRoutine(row) : null; + } + /** * List a user's routines, newest-updated first. Includes paused rows so * the user can see and resume them via the tool. diff --git a/middleware/test/hrRoutineTemplate.integration.test.ts b/middleware/test/hrRoutineTemplate.integration.test.ts index 6fe15d07e..55ad133df 100644 --- a/middleware/test/hrRoutineTemplate.integration.test.ts +++ b/middleware/test/hrRoutineTemplate.integration.test.ts @@ -111,7 +111,7 @@ class InMemoryRoutineStore { cron: input.cron, prompt: input.prompt, channel: input.channel, - conversationRef: input.conversationRef, + conversationRef: input.conversationRef ?? {}, status: 'active', timeoutMs: input.timeoutMs ?? 600_000, createdAt: now, @@ -129,6 +129,18 @@ class InMemoryRoutineStore { return this.rows.get(id) ?? null; } + async getByName( + tenant: string, + userId: string, + name: string, + ): Promise { + return ( + [...this.rows.values()].find( + (r) => r.tenant === tenant && r.userId === userId && r.name === name, + ) ?? null + ); + } + async listForUser(): Promise { return [...this.rows.values()]; } diff --git a/middleware/test/orchestrator/committedToolReporting.test.ts b/middleware/test/orchestrator/committedToolReporting.test.ts new file mode 100644 index 000000000..beebcf095 --- /dev/null +++ b/middleware/test/orchestrator/committedToolReporting.test.ts @@ -0,0 +1,354 @@ +import { describe, it } from 'node:test'; +import { strict as assert } from 'node:assert'; + +import type { + LlmProvider, + LlmRequest, + LlmResponse, + LlmStreamEvent, +} from '@omadia/llm-provider'; +import type { ChatStreamEvent } from '@omadia/channel-sdk'; +import { + NativeToolRegistry, + Orchestrator, + type SessionLogEntry, +} from '@omadia/orchestrator'; + +/** + * Issue #506 — the one-click repro. `chatStreamInner` wraps its whole + * per-turn iteration loop in a single try/catch: an exception thrown by a + * LATER iteration's model call (after an EARLIER iteration's tool call + * already committed a real side effect and yielded a successful + * `tool_result`) used to fall into the same catch-all that reports a bare + * `{ type: 'error' }` — discarding the fact that the action already + * succeeded. These tests exercise the fix: a committed tool result changes + * the catch block's outcome to a `done` event; a genuine failure with no + * prior committed tool result is unaffected. + * + * Review follow-up: the original version of this file never constructed a + * `sessionLogger`, so it couldn't have caught the emergency-`done` path + * skipping `sessionLogger.log()` — the ONE thing every other `done` + * emission site in `chatStreamInner` does before yielding. Every test here + * now builds the orchestrator WITH a recording session logger and asserts + * on what it did (or didn't) receive. + */ + +/** Records every `SessionLogEntry` passed to `log()`, mirroring the stub + * pattern from turnHooks.test.ts (`{ turnExternalId }` return shape). */ +function recordingSessionLogger(): { + sessionLogger: ConstructorParameters[0]['sessionLogger']; + calls: SessionLogEntry[]; +} { + const calls: SessionLogEntry[] = []; + const sessionLogger = { + log: async (entry: SessionLogEntry): Promise<{ turnExternalId: string }> => { + calls.push(entry); + return { turnExternalId: `turn:${entry.scope}:stub` }; + }, + } as unknown as ConstructorParameters[0]['sessionLogger']; + return { sessionLogger, calls }; +} + +interface ScriptedStream { + events: LlmStreamEvent[]; +} + +/** A scripted stream entry that fails immediately (no events at all), the + * way a non-retryable provider error surfaces before any text/tool delta — + * see `streamMessageEvents`'s `forwardedText` retry gate in streaming.ts. */ +interface ThrowingStream { + throws: Error; +} + +const providerCapabilities = { + tools: true, + vision: true, + streaming: true, + promptCaching: true, + forcedToolChoice: true, + parallelToolCalls: true, +} as const; + +/** Mirrors parallelTool.test.ts's fakeStreamProvider, extended with the + * ability to script a call that throws instead of streaming events — the + * shape needed to reproduce a post-tool-dispatch model failure. */ +function fakeStreamProvider( + scripts: Array, +): LlmProvider { + let idx = 0; + const provider = { + id: 'anthropic', + capabilities: providerCapabilities, + complete: async (): Promise => { + throw new Error('fakeStreamProvider: complete() not scripted'); + }, + stream: (_req: LlmRequest): AsyncIterable => { + if (idx >= scripts.length) { + throw new Error( + `fakeStreamProvider: no scripted stream for call ${String(idx + 1)}`, + ); + } + const script = scripts[idx]!; + idx += 1; + if ('throws' in script) { + return { + async *[Symbol.asyncIterator]() { + throw script.throws; + }, + }; + } + return { + async *[Symbol.asyncIterator]() { + for (const ev of script.events) yield ev; + }, + }; + }, + // Non-retryable — the second scripted call's failure must propagate + // straight to `chatStreamInner`'s outer catch, exactly like a genuine + // hard failure would (see isRetryableStreamError for the transient set). + classifyError: () => ({ retryable: false, kind: 'other' as const }), + }; + return provider as unknown as LlmProvider; +} + +function streamWithTools( + toolUses: Array<{ id: string; name: string; input: unknown }>, +): ScriptedStream { + const events: LlmStreamEvent[] = []; + toolUses.forEach((u) => { + events.push( + { type: 'tool_use_start' }, + { type: 'tool_input_delta', text: JSON.stringify(u.input) }, + ); + }); + events.push({ + type: 'final', + response: { + content: toolUses.map((u) => ({ + type: 'tool_call', + id: u.id, + name: u.name, + input: u.input, + })), + finishReason: 'tool_calls', + providerFinishReason: 'tool_use', + model: 'test', + usage: { + inputTokens: 50, + outputTokens: 4, + cacheReadTokens: 0, + cacheWriteTokens: 0, + }, + }, + }); + return { events }; +} + +function buildOrchestrator( + provider: LlmProvider, + registry: NativeToolRegistry, + sessionLogger: ConstructorParameters[0]['sessionLogger'], +): Orchestrator { + return new Orchestrator({ + provider, + model: 'test', + maxTokens: 1024, + maxToolIterations: 5, + domainTools: [], + nativeToolRegistry: registry, + sessionLogger, + }); +} + +const minimalSpec = (name: string): Record => ({ + name, + description: `${name} for testing`, + input_schema: { type: 'object' as const, properties: {}, required: [] }, +}); + +describe('Issue #506 — report success when a tool already committed', () => { + it('ends with `done` (not `error`) when a tool committed in an earlier iteration and a later model call throws', async () => { + const registry = new NativeToolRegistry(); + registry.register('manage_widget', { + handler: async (): Promise => 'widget-created', + // eslint-disable-next-line @typescript-eslint/no-explicit-any + spec: minimalSpec('manage_widget') as any, + }); + + // Iteration 0: model calls the tool, which succeeds and commits. + const stream0 = streamWithTools([ + { id: 'use-1', name: 'manage_widget', input: {} }, + ]); + // Iteration 1: the follow-up model call (generating the natural-language + // confirmation) fails hard — the exact shape a mid-stream, non-retryable + // provider error takes, or what's left once internal retries in + // streamMessageEvents are exhausted. + const stream1: ThrowingStream = { + throws: Object.assign(new Error('boom: provider hard-failed'), { + status: 400, + }), + }; + const provider = fakeStreamProvider([stream0, stream1]); + const { sessionLogger, calls } = recordingSessionLogger(); + const orchestrator = buildOrchestrator(provider, registry, sessionLogger); + + const events: ChatStreamEvent[] = []; + for await (const ev of orchestrator.chatStream({ + userMessage: 'create a widget', + sessionScope: 'sess-506-committed', + })) { + events.push(ev); + } + + const errorEvents = events.filter((e) => e.type === 'error'); + const doneEvents = events.filter((e) => e.type === 'done'); + assert.equal( + errorEvents.length, + 0, + `expected no error event, got ${JSON.stringify(errorEvents)}`, + ); + assert.equal(doneEvents.length, 1, 'expected exactly one done event'); + + const done = doneEvents[0]; + assert.ok(done && done.type === 'done'); + if (done && done.type === 'done') { + assert.match(done.answer, /manage_widget/); + assert.match(done.answer, /completed successfully/i); + assert.equal(done.toolCalls, 1); + // Two iterations were entered (0 and 1) before the failure. + assert.equal(done.iterations, 2); + } + + // Review follow-up: the emergency-`done` path must persist the + // exchange exactly like every other `done` emission site does — + // otherwise the committed `manage_widget` call is invisible to the + // next turn and the model could re-invoke it, duplicating the side + // effect. Assert the logger actually ran, with fields matching what + // was yielded to the caller. + assert.equal(calls.length, 1, 'expected sessionLogger.log to be called once'); + const logged = calls[0]; + assert.ok(logged); + if (logged) { + assert.equal(logged.scope, 'sess-506-committed'); + assert.equal(logged.userMessage, 'create a widget'); + assert.equal(logged.assistantAnswer, done && done.type === 'done' ? done.answer : undefined); + assert.equal(logged.toolCalls, 1); + assert.equal(logged.iterations, 2); + } + }); + + it('still ends with `error` when no tool committed before the failure (genuine failure, unchanged behavior)', async () => { + const registry = new NativeToolRegistry(); + + // Iteration 0: the very first model call fails hard, before any tool ran. + const stream0: ThrowingStream = { + throws: Object.assign(new Error('boom: provider hard-failed'), { + status: 400, + }), + }; + const provider = fakeStreamProvider([stream0]); + const { sessionLogger, calls } = recordingSessionLogger(); + const orchestrator = buildOrchestrator(provider, registry, sessionLogger); + + const events: ChatStreamEvent[] = []; + for await (const ev of orchestrator.chatStream({ + userMessage: 'do something', + sessionScope: 'sess-506-genuine-failure', + })) { + events.push(ev); + } + + const errorEvents = events.filter((e) => e.type === 'error'); + const doneEvents = events.filter((e) => e.type === 'done'); + assert.equal(doneEvents.length, 0, 'expected no done event'); + assert.equal(errorEvents.length, 1, 'expected exactly one error event'); + const error = errorEvents[0]; + assert.ok(error && error.type === 'error'); + if (error && error.type === 'error') { + assert.match(error.message, /boom: provider hard-failed/); + } + + // A genuine failure with nothing committed must NOT persist a session + // log entry — unchanged behavior, same as every other `error` emission + // site in chatStreamInner. + assert.equal( + calls.length, + 0, + 'expected sessionLogger.log NOT to be called on a genuine failure', + ); + }); + + it('reports done even when a later intended action never ran (accepted tradeoff, see code comment)', async () => { + // This test PINS the maintainer-reviewed, deliberate tradeoff documented + // on `committedToolNames` and the catch-block done-vs-error branch in + // orchestrator.ts — it does NOT assert that this behavior is correct in + // all cases. A read-only-style tool (`list_routines`) succeeds in + // iteration 0. Iteration 1's model call — which, had it succeeded, would + // have requested a SECOND, different, mutating tool call (e.g. a + // `manage_routine` create) — throws before it can request that second + // tool call at all. The committed-tool tracking is tool-agnostic: it + // cannot distinguish "a harmless read succeeded" from "the consequential + // write the user actually wanted never ran." The turn still reports + // `done`, naming only the read-only tool that actually committed — this + // is the accepted residual risk, not a guarantee that the user's + // intended action happened. + const registry = new NativeToolRegistry(); + registry.register('list_routines', { + handler: async (): Promise => 'routine-a, routine-b', + // eslint-disable-next-line @typescript-eslint/no-explicit-any + spec: minimalSpec('list_routines') as any, + }); + + // Iteration 0: model calls the read-only tool, which succeeds and + // commits (per the generic, tool-agnostic tracking). + const stream0 = streamWithTools([ + { id: 'use-1', name: 'list_routines', input: {} }, + ]); + // Iteration 1: the model call that would have gone on to request a + // second, mutating tool call (never scripted here — it never gets that + // far) fails hard instead. + const stream1: ThrowingStream = { + throws: Object.assign( + new Error('boom: provider hard-failed before requesting the mutating tool'), + { status: 400 }, + ), + }; + const provider = fakeStreamProvider([stream0, stream1]); + const { sessionLogger, calls } = recordingSessionLogger(); + const orchestrator = buildOrchestrator(provider, registry, sessionLogger); + + const events: ChatStreamEvent[] = []; + for await (const ev of orchestrator.chatStream({ + userMessage: 'list my routines, then create a new one', + sessionScope: 'sess-506-later-action-skipped', + })) { + events.push(ev); + } + + const errorEvents = events.filter((e) => e.type === 'error'); + const doneEvents = events.filter((e) => e.type === 'done'); + assert.equal( + errorEvents.length, + 0, + `expected no error event (accepted tradeoff), got ${JSON.stringify(errorEvents)}`, + ); + assert.equal(doneEvents.length, 1, 'expected exactly one done event'); + + const done = doneEvents[0]; + assert.ok(done && done.type === 'done'); + if (done && done.type === 'done') { + // Only the read-only tool that actually committed is named — the + // never-requested mutating tool is (correctly, per the generic + // tracking) absent from the answer. Nothing here claims the intended + // create actually happened. + assert.match(done.answer, /list_routines/); + assert.equal(done.toolCalls, 1); + assert.equal(done.iterations, 2); + } + + // The emergency-done path still persists the (partial) exchange, same + // as the single-tool case above — that part of the behavior is not + // being challenged by this test. + assert.equal(calls.length, 1, 'expected sessionLogger.log to be called once'); + }); +}); diff --git a/middleware/test/routineRunner.test.ts b/middleware/test/routineRunner.test.ts index db939a842..880d6fcae 100644 --- a/middleware/test/routineRunner.test.ts +++ b/middleware/test/routineRunner.test.ts @@ -26,6 +26,7 @@ import { type JobSchedulerLike, type OrchestratorLike, } from '../src/plugins/routines/routineRunner.js'; +import { RoutineNameConflictError } from '../src/plugins/routines/routineStore.js'; import { routineTurnContext } from '../src/plugins/routines/routineTurnContext.js'; import { turnContext } from '@omadia/orchestrator'; import type { RoutineOutputTemplate } from '../src/plugins/routines/routineOutputTemplate.js'; @@ -96,6 +97,18 @@ class InMemoryRoutineStore implements RoutineStore { // pg pool field is required by TS — this stub just satisfies the // structural shape the runner relies on (duck-typed via the import). async create(input: CreateRoutineInput): Promise { + // Mirror the real store's `routines_user_name_unique` constraint so + // tests can exercise the runner's conflict-reconciliation path + // (issue #506) without a real Postgres pool. + const collision = [...this.rows.values()].find( + (r) => + r.tenant === input.tenant && + r.userId === input.userId && + r.name === input.name, + ); + if (collision) { + throw new RoutineNameConflictError(input.name); + } const id = `routine-${this.nextId++}`; const now = new Date(); const routine: Routine = { @@ -106,7 +119,7 @@ class InMemoryRoutineStore implements RoutineStore { cron: input.cron, prompt: input.prompt, channel: input.channel, - conversationRef: input.conversationRef, + conversationRef: input.conversationRef ?? {}, status: 'active', timeoutMs: input.timeoutMs ?? 600_000, createdAt: now, @@ -124,6 +137,18 @@ class InMemoryRoutineStore implements RoutineStore { return this.rows.get(id) ?? null; } + async getByName( + tenant: string, + userId: string, + name: string, + ): Promise { + return ( + [...this.rows.values()].find( + (r) => r.tenant === tenant && r.userId === userId && r.name === name, + ) ?? null + ); + } + async listForUser(tenant: string, userId: string): Promise { return [...this.rows.values()].filter( (r) => r.tenant === tenant && r.userId === userId, @@ -372,6 +397,34 @@ describe('RoutineRunner — createRoutine', () => { ); }); + // Issue #506 (reviewer-confirmed gap) — the quota check must not run + // before the reconciliation check for a user already at capacity. A user + // at `maxActivePerUser` who retries the identical `createRoutine` call + // that already brought them there (e.g. their turn's own confirmation + // was dropped) must reconcile to the existing row, not be told they're + // out of quota — that's the exact false-negative issue #506 was filed + // over, just re-surfaced as a different exception type. + it('reconciles a same-request retry even when the user is already at quota', async () => { + const h = makeHarness({ maxActivePerUser: 1 }); + const first = await h.runner.createRoutine(baseInput); + const retry = await h.runner.createRoutine(baseInput); + assert.equal(retry.id, first.id, 'retry must resolve to the original row'); + assert.equal(h.store.rows.size, 1, 'no duplicate row was created'); + }); + + // The quota gate must still apply to a genuinely new routine request — + // reconciliation is only for retries of an already-existing, already- + // active routine. A different `name` is a real new-routine request and + // must not silently bypass the quota check. + it('still enforces the per-user quota for a genuinely new routine name', async () => { + const h = makeHarness({ maxActivePerUser: 1 }); + await h.runner.createRoutine(baseInput); + await assert.rejects( + () => h.runner.createRoutine({ ...baseInput, name: 'a-different-name' }), + RoutineQuotaExceededError, + ); + }); + it('rolls the row back when scheduler.register throws', async () => { // Production: JobScheduler validates cron via croner and throws // JobValidationError on malformed input. We simulate that here so the @@ -390,6 +443,202 @@ describe('RoutineRunner — createRoutine', () => { 'delete() should be invoked exactly once on rollback', ); }); + + // Issue #506 — a retry that lands on the same (tenant, user, name) unique + // key must reconcile against the row that already exists rather than + // reporting a failure (the row from the FIRST call already committed; + // only the confirmation leg back to the caller was ever in doubt). + it('reconciles a same-request retry instead of reporting a conflict', async () => { + const h = makeHarness(); + const first = await h.runner.createRoutine(baseInput); + const retry = await h.runner.createRoutine(baseInput); + assert.equal(retry.id, first.id, 'retry must resolve to the original row'); + assert.equal(h.store.rows.size, 1, 'no duplicate row was created'); + // Only one scheduler registration — the reconciled retry does not + // re-register (would throw JobAlreadyRegisteredError against the real + // scheduler if it tried). + assert.equal(h.scheduler.list().length, 1); + }); + + it('still reports a conflict when the retry disagrees with the existing routine', async () => { + const h = makeHarness(); + await h.runner.createRoutine(baseInput); + await assert.rejects( + () => + h.runner.createRoutine({ ...baseInput, prompt: 'Sag etwas anderes' }), + RoutineNameConflictError, + ); + assert.equal(h.store.rows.size, 1, 'the conflicting attempt created nothing'); + }); + + // Reviewer-confirmed bug fix: `outputTemplate` is an independently-settable + // object field on both `Routine` and `CreateRoutineInput` (Phase C output + // templates) and must be compared, not ignored. A create that agrees on + // cron/prompt/channel/timeoutMs but asks for a different `outputTemplate` + // is the caller changing the structured-output template on an existing + // schedule, not a retry — silently reconciling to the old row would + // discard the caller's new template while reporting success. + it('still reports a conflict when the retry differs only in outputTemplate', async () => { + const h = makeHarness(); + const template: RoutineOutputTemplate = { + format: 'markdown', + sections: [ + { + kind: 'static-markdown', + text: 'Original', + }, + ], + }; + const otherTemplate: RoutineOutputTemplate = { + format: 'markdown', + sections: [ + { + kind: 'static-markdown', + text: 'Different', + }, + ], + }; + await h.runner.createRoutine({ ...baseInput, outputTemplate: template }); + await assert.rejects( + () => + h.runner.createRoutine({ + ...baseInput, + outputTemplate: otherTemplate, + }), + RoutineNameConflictError, + ); + assert.equal(h.store.rows.size, 1, 'the conflicting attempt created nothing'); + }); + + it('reconciles a retry whose outputTemplate is structurally identical', async () => { + const h = makeHarness(); + const template: RoutineOutputTemplate = { + format: 'markdown', + sections: [ + { + kind: 'static-markdown', + text: 'Same', + }, + ], + }; + const first = await h.runner.createRoutine({ + ...baseInput, + outputTemplate: template, + }); + // A structurally-equal but distinct object — not the same reference — + // must still reconcile; the comparison is deep, not reference equality. + const retry = await h.runner.createRoutine({ + ...baseInput, + outputTemplate: { + format: 'markdown', + sections: [ + { + kind: 'static-markdown', + text: 'Same', + }, + ], + }, + }); + assert.equal(retry.id, first.id, 'retry must resolve to the original row'); + assert.equal(h.store.rows.size, 1, 'no duplicate row was created'); + }); + + // Reviewer-confirmed bug fix: `conversationRef` is caller-specified on the + // cold-start outreach path (`ManageRoutineTool.handleCreate` resolves it + // from `targetEmail` via `buildEmailColdStartTarget` before calling + // `createRoutine`), so it must be compared like `outputTemplate`, not + // ignored. A create that agrees on cron/prompt/channel/timeoutMs/ + // outputTemplate but resolves a DIFFERENT `conversationRef` (e.g. a + // different `targetEmail`) is a genuine new request aimed at a different + // recipient — silently reconciling to the existing row would report + // "created" while the new recipient never gets set up and the routine + // keeps messaging the original one. + it('still reports a conflict when the retry resolves a different conversationRef', async () => { + const h = makeHarness(); + await h.runner.createRoutine({ + ...baseInput, + conversationRef: { conversation: { id: 'conv-alice' } }, + }); + await assert.rejects( + () => + h.runner.createRoutine({ + ...baseInput, + conversationRef: { conversation: { id: 'conv-bob' } }, + }), + RoutineNameConflictError, + ); + assert.equal(h.store.rows.size, 1, 'the conflicting attempt created nothing'); + }); + + it('reconciles a retry whose conversationRef is structurally identical', async () => { + const h = makeHarness(); + const first = await h.runner.createRoutine({ + ...baseInput, + conversationRef: { conversation: { id: 'conv-alice' } }, + }); + // A structurally-equal but distinct object — not the same reference — + // must still reconcile; the comparison is deep, not reference equality. + const retry = await h.runner.createRoutine({ + ...baseInput, + conversationRef: { conversation: { id: 'conv-alice' } }, + }); + assert.equal(retry.id, first.id, 'retry must resolve to the original row'); + assert.equal(h.store.rows.size, 1, 'no duplicate row was created'); + }); + + // Reviewer-confirmed bug fix: `store.create()` normalizes an omitted + // `conversationRef` to `{}` before persisting (routineStore.ts stores + // `JSON.stringify(input.conversationRef ?? {})`, and reads it back the + // same way), but `isSameRoutineRequest` used to compare the stored `{}` + // against the RAW retry input with no equivalent `?? {}` normalization — + // unlike `timeoutMs` and `outputTemplate`, which already apply the same + // default the store itself uses. A genuine retry of the ordinary + // (non-cold-start) create path — where `conversationRef` is legitimately + // `undefined` both times — must still reconcile to the existing row + // rather than falling through to `RoutineNameConflictError`. + it('reconciles a retry whose conversationRef is omitted both times', async () => { + const h = makeHarness(); + const first = await h.runner.createRoutine({ + ...baseInput, + conversationRef: undefined, + }); + const retry = await h.runner.createRoutine({ + ...baseInput, + conversationRef: undefined, + }); + assert.equal(retry.id, first.id, 'retry must resolve to the original row'); + assert.equal(h.store.rows.size, 1, 'no duplicate row was created'); + }); + + // Reviewer-confirmed bug fix: reconciliation must not fire against a + // paused (or otherwise non-active) existing row, even when every other + // field matches byte-for-byte. A paused routine is not "my own in-flight + // retry" — it's a separate, deliberate create call that happens to share + // a name with something the user already paused. Silently returning the + // paused row would report `status: 'paused'` as a successful create with + // no active schedule, which is a worse instance of the exact + // false-negative/false-positive problem issue #506 was filed to fix. + it('still reports a conflict when the existing same-name routine is paused', async () => { + const h = makeHarness(); + const first = await h.runner.createRoutine(baseInput); + await h.runner.pauseRoutine(first.id); + + await assert.rejects( + () => h.runner.createRoutine(baseInput), + RoutineNameConflictError, + ); + assert.equal(h.store.rows.size, 1, 'no new row was created'); + assert.equal( + h.store.rows.get(first.id)?.status, + 'paused', + 'the existing paused row must be left untouched', + ); + assert.equal( + h.scheduler.list().length, + 0, + 'no scheduler registration should have happened', + ); + }); }); describe('RoutineRunner — start (boot scan)', () => {