From 6f1d67d71b2d7206596286e213e4bfba7abaf7ac Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Wed, 26 Aug 2026 07:27:43 +0200 Subject: [PATCH] feat(api): flag re-executed investigation lanes; correct the TTFT comment A lane step whose result is lost to a retry boundary is re-run by the workflow engine on its own schedule - minutes later, with no failure recorded anywhere. The re-execution's spans were indistinguishable from new work, so a session view showed a silent multi-minute hole followed by what looked like fresh lanes. The step now reads the lane row before claiming it (past "queued" means a prior execution ran) and stamps maple.hypothesis.rerun on the lane's root span. Also corrects the annotateModelCallTiming comment: the openai-chat protocol emits no step-start, and role-only chunks, keep-alives and hidden reasoning produce no events, so TTFT is measured to the first token-bearing frame - which on reasoning models spans the silent reasoning phase. The behavior is intended; the comment claimed first-byte semantics it never had. --- apps/api/src/chat/loop/genai.ts | 13 +++++--- .../InvestigationFanoutWorkflow.run.test.ts | 31 +++++++++++++++++++ .../InvestigationFanoutWorkflow.run.ts | 22 +++++++++++++ apps/api/src/workflows/hypothesis-agent.ts | 4 +++ 4 files changed, 66 insertions(+), 4 deletions(-) diff --git a/apps/api/src/chat/loop/genai.ts b/apps/api/src/chat/loop/genai.ts index 75038eba7..b3c4146b4 100644 --- a/apps/api/src/chat/loop/genai.ts +++ b/apps/api/src/chat/loop/genai.ts @@ -449,7 +449,7 @@ export const annotateModelCallEnd = (events: ReadonlyArray): Effect.Ef * The span's wall clock covers the whole stream lifetime — including the SSE * frames and durable writes that *consume* the model's output — so it * systematically overstates the model. These two attributes are the honest - * numbers: request start → first provider frame (`time_to_first_chunk`, + * numbers: request start → first token-bearing event (`time_to_first_chunk`, * seconds, the key the session view's occupancy bar reads), and request start * → terminal event (model duration, milliseconds). */ @@ -460,9 +460,14 @@ export interface ModelCallTiming { /** * Annotate the current model-call span as events arrive: TTFT on the first - * provider frame (`step-start` is emitted while processing that frame, never - * before the network, so the first event of any type is the first byte), the - * model's own duration on the terminal event. + * event, the model's own duration on the terminal event. + * + * The first event is the first *token-bearing* frame, not the first byte: the + * openai-chat protocol emits no `step-start`, and role-only chunks, keep-alive + * comments and hidden reasoning produce no events at all. So on a reasoning + * model TTFT spans the whole silent reasoning phase — deliberately, since that + * is when the first token actually reached this side of the wire. On + * tool-call-only steps this puts TTFT near the span's full duration. */ export const annotateModelCallTiming = (timing: ModelCallTiming, event: LLMEvent): Effect.Effect => Effect.suspend(() => { diff --git a/apps/api/src/workflows/InvestigationFanoutWorkflow.run.test.ts b/apps/api/src/workflows/InvestigationFanoutWorkflow.run.test.ts index fe35578fd..1b8f59da2 100644 --- a/apps/api/src/workflows/InvestigationFanoutWorkflow.run.test.ts +++ b/apps/api/src/workflows/InvestigationFanoutWorkflow.run.test.ts @@ -234,6 +234,37 @@ describe("runInvestigationFanout", () => { } }) + /** + * The engine can re-run a lane step whose result was lost to a retry boundary, + * minutes later and with no failure recorded anywhere. The second execution + * must know it is one, so its span does not read as new work after a silent + * gap in the session view. + */ + it("flags a re-executed lane step as a rerun", async () => { + const seen: Array<{ id: string; rerun: boolean }> = [] + const doubleStep: WorkflowStepLike = { + do: (async (name: string, configOrCb: unknown, cb?: () => Promise) => { + const callback = cb ?? (configOrCb as () => Promise) + const first = await callback() + return name.startsWith("hypothesis-") ? callback() : first + }) as WorkflowStepLike["do"], + } + await runInvestigationFanout( + env, + { payload: harness.payload }, + doubleStep, + baseDeps({ + invokeHypothesis: async ({ hypothesis, rerun }) => { + seen.push({ id: hypothesis.id, rerun }) + return hypothesisOutput(hypothesis.id) + }, + }), + ) + for (const id of HYPOTHESIS_IDS) { + expect(seen.filter((entry) => entry.id === id).map((entry) => entry.rerun)).toEqual([false, true]) + } + }) + /** * The reservation is made before the planner runs, so it is deliberately high. * Left unreconciled, an org's daily pass budget drains at the ceiling rather diff --git a/apps/api/src/workflows/InvestigationFanoutWorkflow.run.ts b/apps/api/src/workflows/InvestigationFanoutWorkflow.run.ts index 1486d4f33..52c0e9f0e 100644 --- a/apps/api/src/workflows/InvestigationFanoutWorkflow.run.ts +++ b/apps/api/src/workflows/InvestigationFanoutWorkflow.run.ts @@ -308,6 +308,9 @@ export interface InvokeHypothesisInput { readonly deadlineAtMs: number /** True on the collapsed path: answer with a full diagnosis, not a candidate. */ readonly solo: boolean + /** True when this lane's row shows a prior execution — the step re-ran after + * its result was lost to a retry boundary. */ + readonly rerun: boolean readonly runtime: SharedRuntime } @@ -363,6 +366,7 @@ const invokeHypothesis = async (input: InvokeHypothesisInput): Promise => { const startedAt = clock() + // A lane row past "queued" means this callback already ran and its step + // result was lost to a retry boundary (an engine reschedule can land + // minutes later, with no failure recorded anywhere). Stamped on the + // lane's span so a session view can tell a re-run from new work. + const prior = await dbStep((db) => + db + .select({ status: investigationLensRuns.status }) + .from(investigationLensRuns) + .where( + and( + eq(investigationLensRuns.investigationId, idTyped), + eq(investigationLensRuns.attempt, attempt), + eq(investigationLensRuns.lensId, hypothesis.id), + ), + ), + ) + const rerun = prior[0] !== undefined && prior[0].status !== "queued" await dbStep((db) => db .update(investigationLensRuns) @@ -879,6 +900,7 @@ async function runWithDb( snapshot, deadlineAtMs: hypothesisDeadlineAtMs, solo: planned.collapsed, + rerun, runtime, }) const finishedAt = clock() diff --git a/apps/api/src/workflows/hypothesis-agent.ts b/apps/api/src/workflows/hypothesis-agent.ts index a7a9ae247..b2aed7d19 100644 --- a/apps/api/src/workflows/hypothesis-agent.ts +++ b/apps/api/src/workflows/hypothesis-agent.ts @@ -31,6 +31,8 @@ export interface HypothesisAgentInput { readonly tenant: TenantContext /** Wall-clock budget; the turn spends one last step submitting past it. */ readonly deadlineAtMs: number + /** The workflow step re-ran after its result was lost to a retry boundary. */ + readonly rerun: boolean } export interface HypothesisAgentOutput { @@ -54,6 +56,7 @@ export const runHypothesisAgent = Effect.fn("investigation.hypothesis")(function yield* Effect.annotateCurrentSpan({ "maple.investigation.id": input.investigationId, "maple.hypothesis.id": input.hypothesis.id, + ...(input.rerun ? { "maple.hypothesis.rerun": true } : undefined), }) const pass = yield* runAgentPass({ @@ -125,6 +128,7 @@ export const runSoloHypothesisAgent = Effect.fn("investigation.solo")(function* "maple.investigation.id": input.investigationId, "maple.hypothesis.id": input.hypothesis.id, "maple.investigation.collapsed": true, + ...(input.rerun ? { "maple.hypothesis.rerun": true } : undefined), }) const pass = yield* runAgentPass({