From 1701c48f177b3832b08b8b56bd2050c4b44e8ea4 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 20 Oct 2026 20:28:27 -0500 Subject: [PATCH] fix: retain chat origin through long continuation chains --- .../chat-monitor-continuation.test.ts | 39 +++++++++++++++++++ server/src/services/issues.ts | 5 ++- 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/server/src/services/chat-monitor-continuation.test.ts b/server/src/services/chat-monitor-continuation.test.ts index 0f3281c067e..26552652951 100644 --- a/server/src/services/chat-monitor-continuation.test.ts +++ b/server/src/services/chat-monitor-continuation.test.ts @@ -64,6 +64,28 @@ if (!support.supported) console.warn(`Monitor publication database unavailable: expect(publications[0]).toMatchObject({ conversationId: f.conversation.id, payload: { progressState: "working" } }); }); + it("retains the verified conversation after more than sixteen monitor admissions", async () => { + const f = await fixture(); + let previous = f.successor; + for (let index = 0; index < 20; index += 1) { + const nextCheckAt = new Date(Date.now() + 120_000 + index * 1000).toISOString(); + await db.update(heartbeatRuns).set({ status: "succeeded", resultJson: { semanticToolReceipts: { + schedule: { operationId: "set_task_monitor", result: { taskId: f.issue.id, monitor: { status: "scheduled", nextCheckAt } } }, + } } }).where(eq(heartbeatRuns.id, previous.id)); + const [next] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + nativeIssueId: f.issue.id, runtimeMode: "native", status: "running", responsibleUserId: "requester", + contextSnapshot: { source: "issue.monitor", issueId: f.issue.id, nextCheckAt }, + }).returning(); + await db.insert(agentWakeupRequests).values({ companyId: f.company.id, agentId: f.agent.id, + source: "automation", reason: "issue_monitor_due", status: "claimed", runId: next.id, + idempotencyKey: `issue-monitor:${f.issue.id}:${nextCheckAt}`, payload: { issueId: f.issue.id, nextCheckAt }, + }); + previous = next; + } + expect(await f.bindings(previous.id)).toEqual(await f.bindings(f.source.id)); + expect(await f.bindings(previous.id)).toHaveLength(1); + }); + it.each(["receipt", "wake", "requester", "agent", "schedule", "task"])("rejects broken %s lineage even with a claimed source id", async fault => { const f = await fixture(); await db.update(heartbeatRuns).set({ contextSnapshot: { ...f.successor.contextSnapshot, sourceRunId: f.source.id } }).where(eq(heartbeatRuns.id, f.successor.id)); @@ -84,6 +106,23 @@ if (!support.supported) console.warn(`Monitor publication database unavailable: expect(await f.bindings()).toEqual([]); }); + it("rejects a cycle of otherwise matching monitor receipts", async () => { + const f = await fixture(); + const createdAt = new Date(Date.now() - 1000); + await db.update(heartbeatRuns).set({ createdAt }).where(eq(heartbeatRuns.companyId, f.company.id)); + const nextCheckAt = new Date(Date.now() + 120_000).toISOString(); + await db.update(heartbeatRuns).set({ contextSnapshot: { source: "issue.monitor", issueId: f.issue.id } }) + .where(eq(heartbeatRuns.id, f.source.id)); + await db.update(heartbeatRuns).set({ resultJson: { semanticToolReceipts: { + schedule: { operationId: "set_task_monitor", result: { taskId: f.issue.id, monitor: { status: "scheduled", nextCheckAt } } }, + } } }).where(eq(heartbeatRuns.id, f.successor.id)); + await db.insert(agentWakeupRequests).values({ companyId: f.company.id, agentId: f.agent.id, + source: "automation", reason: "issue_monitor_due", status: "claimed", runId: f.source.id, + idempotencyKey: `issue-monitor:${f.issue.id}:${nextCheckAt}`, payload: { issueId: f.issue.id, nextCheckAt }, + }); + expect(await f.bindings()).toEqual([]); + }); + it("does not turn a yielded successful provider run into a completion", async () => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "succeeded", resultJson: { diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 6146de965d3..05cf67a8e18 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1154,7 +1154,10 @@ export async function resolveChatOriginPublicationBindings( let contextSnapshot: Record | null = null; let lineageAgentId: string | null = null; const visitedRunIds = new Set(); - for (let depth = 0; depth < 16; depth += 1) { + // Every hop needs its own durable authorization receipt. A task may continue + // many times; an arbitrary hop limit must not silently discard its replies. + // The visited set rejects cycles while keeping valid long lineages readable. + while (true) { if (visitedRunIds.has(originRunId)) return []; visitedRunIds.add(originRunId); const run = await dbOrTx