diff --git a/server/src/services/execution-continuation-provenance.test.ts b/server/src/services/execution-continuation-provenance.test.ts index fd294b88ffd..ce9a12edd6b 100644 --- a/server/src/services/execution-continuation-provenance.test.ts +++ b/server/src/services/execution-continuation-provenance.test.ts @@ -98,11 +98,22 @@ const support = await getEmbeddedPostgresTestSupport(); expect(JSON.stringify(envelope)).not.toContain("Unrelated"); }); - it.each(["retryOfRunId", "previousRunId", "interruptedRunId"])("rejects explicit %s history from another task", async (key) => { + it.each(["retryOfRunId", "previousRunId"])("rejects explicit %s history from another task", async (key) => { const f = await fixture(); await expect(f.build({ [key]: f.producerRunId })).rejects.toThrow("continuation_source_context_missing"); }); + it("degrades a stale interruptedRunId hint from another task and keeps producer attribution", async () => { + // interruptedRunId is a best-effort hint, not explicit user history: when + // it names another task's run the wake continues without that resume + // source, and an interaction wake falls back to its producer. + const f = await fixture(); + const envelope = await f.build({ interruptedRunId: f.producerRunId }); + expect(envelope.trigger.sourceRunId).toBe(f.producerRunId); + expect(envelope.completedWork).toBeNull(); + expect(envelope).not.toHaveProperty("interruptedRunId"); + }); + it("retains the explicit user continuation authorization requirement", async () => { const f = await fixture(); await expect(f.build({ explicitUserContinuation: { previousRunId: f.producerRunId } })) diff --git a/server/src/services/execution-continuation.test.ts b/server/src/services/execution-continuation.test.ts index 8f027562972..834bf515ee7 100644 --- a/server/src/services/execution-continuation.test.ts +++ b/server/src/services/execution-continuation.test.ts @@ -193,14 +193,41 @@ const support = await getEmbeddedPostgresTestSupport(); } }); - it("rejects handoff history from a different task", async () => { + it("degrades a stale interrupted-run hint from a different task instead of failing the wake", async () => { + // Cross-issue reassigns can leave interruptedRunId pointing at another + // task's run. The hint is best-effort: the wake must continue without a + // resume source instead of throwing continuation_source_context_missing. const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: randomUUID() } }).where(eq(heartbeatRuns.id, runId)); try { - await expectMissingContinuationContext( - () => buildExecutionContinuation({ db, companyId, issueId, agentId, - context: { interruptedRunId: runId }, summary: null, exposeLowTrustRaw: false }), - ); + const envelope = await buildExecutionContinuation({ db, companyId, issueId, agentId, + context: { interruptedRunId: runId, wakeReason: "issue_assigned" }, summary: null, exposeLowTrustRaw: false }); + expect(envelope.trigger.sourceRunId).toBeNull(); + expect(envelope).not.toHaveProperty("interruptedRunId"); + expect(envelope.objective).toBe("Focus the Gmail summary on launch decisions."); + } finally { + await db.update(heartbeatRuns).set({ contextSnapshot: source.contextSnapshot }).where(eq(heartbeatRuns.id, runId)); + } + }); + + it("degrades an interrupted-run hint whose run row no longer exists", async () => { + const missingRunId = randomUUID(); + const envelope = await buildExecutionContinuation({ db, companyId, issueId, agentId, + context: { interruptedRunId: missingRunId, wakeReason: "issue_assigned" }, summary: null, exposeLowTrustRaw: false }); + expect(envelope.trigger.sourceRunId).toBeNull(); + expect(envelope).not.toHaveProperty("interruptedRunId"); + }); + + it("still rejects an explicit user resume from a different task", async () => { + // Fail-closed stays in force for explicit user resume history; only the + // best-effort interruptedRunId hint degrades. + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: randomUUID() } }).where(eq(heartbeatRuns.id, runId)); + try { + await expect( + buildExecutionContinuation({ db, companyId, issueId, agentId, + context: { explicitUserContinuation: { previousRunId: runId }, wakeReason: "user_resume" }, summary: null, exposeLowTrustRaw: false }), + ).rejects.toThrow("continuation_user_authorization_missing"); } finally { await db.update(heartbeatRuns).set({ contextSnapshot: source.contextSnapshot }).where(eq(heartbeatRuns.id, runId)); } diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index 53f1f479817..c44ce6ee325 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -197,25 +197,44 @@ export async function buildExecutionContinuation(input: { ); const explicitContinuation = object(input.context.explicitUserContinuation); const explicitUserSource = string(explicitContinuation.previousRunId); + const interruptedSource = string(input.context.interruptedRunId); const resumeSourceRunId = explicitUserSource ?? string(input.context.retryOfRunId) ?? string(input.context.previousRunId) ?? - string(input.context.interruptedRunId); + interruptedSource; const producerRunId = triggerInteraction?.sourceRunId ?? null; - const sourceRunId = resumeSourceRunId ?? producerRunId; + const requestedSourceRunId = resumeSourceRunId ?? producerRunId; const loadRun = async (id: string) => (await db .select({ context: heartbeatRuns.contextSnapshot, result: heartbeatRuns.resultJson }) .from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, id))) )[0]; - const candidate = sourceRunId ? await loadRun(sourceRunId) : null; + const candidate = requestedSourceRunId ? await loadRun(requestedSourceRunId) : null; // An interaction producer is provenance. Explicit resume history must still // belong to this task, and only task-scoped content can enter the envelope. const sourceRun = object(candidate?.context).issueId === issueId ? candidate : null; - if ((sourceRunId && !candidate) || (resumeSourceRunId && !sourceRun)) + // interruptedRunId is a best-effort hint written when a run is interrupted. + // After a cross-issue reassign it can name another task's run or a deleted + // row. Unlike an explicit user resume or a retry chain, a hint that does + // not resolve to this task must not fail the whole wake: degrade to "no + // resume source" and continue. + const staleInterruptedSource = + interruptedSource !== null + && resumeSourceRunId === interruptedSource + && explicitUserSource === null + && string(input.context.retryOfRunId) === null + && string(input.context.previousRunId) === null + && sourceRun === null; + // A degraded hint falls back to the interaction producer so interaction + // wakes keep their source attribution. + const sourceRunId = staleInterruptedSource ? (producerRunId ?? null) : requestedSourceRunId; + if ( + !staleInterruptedSource + && ((requestedSourceRunId && !candidate) || (resumeSourceRunId && !sourceRun)) + ) throw new Error(explicitUserSource ? "continuation_user_authorization_missing" : "continuation_source_context_missing"); - const producer = producerRunId === sourceRunId ? candidate + const producer = producerRunId === requestedSourceRunId ? candidate : producerRunId ? await loadRun(producerRunId) : null; if (producerRunId && !producer) throw new Error("continuation_source_context_missing"); const producerIssueId = string(object(producer?.context).issueId); @@ -431,7 +450,7 @@ export async function buildExecutionContinuation(input: { if (!predecessor || !authorization || explicitUserSource !== sourceRunId) throw new Error("continuation_user_authorization_missing"); } - const interruptedRunId = explicitUserSource ?? string(input.context.interruptedRunId) ?? (lastTerminal && lastTerminal.status !== "succeeded" && + const interruptedRunId = explicitUserSource ?? (staleInterruptedSource ? undefined : interruptedSource) ?? (lastTerminal && lastTerminal.status !== "succeeded" && (hasConversationContinuationPolicy(lastTerminal.result) || lastTerminal.status === "interrupted" || lastTerminal.errorCode === "process_lost") ? lastTerminal.id : undefined);