diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index a54d9b1898b..ec32eeba90f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1616,3 +1616,19 @@ review blocker is resolved merely by pushing inherited changes. No PRs merged. Origin/master refreshed to e335265568 (task-run UI polling removal #17608). Full base reconciliation, remaining recurring-work slices, milestone startup wiring, fresh reviews, and all latest-head CI gates remain required. + + +Slice 28 Telegram migration now registers affected issue intents in its owning +transaction before reactivating/rebinding conversations. The missing wake was +reproduced first. All 92 milestone/resource/governed tests, eight existing +Telegram migration cases, server typecheck and server build pass. Duplicate +migration and unchanged inventory remain quiet. See the milestone audit for +failure and verification logs. Inbound activation and remaining producer audits +still gate production wiring; no recurring projector has been removed yet. + +Fresh pre-push gates: #17606 at b634353059 has passing checks; #17572 at +8815dfad38 and #17490 at 31bcd19888 still fail E2E shard 3 and aggregate. +Three local cancellation diagnostic runs passed, including full projects coverage +and a database-only delay, without proving the CI deadlock's competing owner. +Those failures, parent security/deletion findings, all other original slices, +fresh remote review and reconciliation with master 7ebf98eafd remain open. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index e6f7853d194..b431518c2f7 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -1152,3 +1152,38 @@ Slack waiting-to-active helpers preserve selector eligibility (both states are already admitted). Email uses explicit publication mode, which the lifecycle selector and lineage resolver exclude. These dispositions narrow the remaining configuration audit; they do not authorize removing the periodic scan yet. + + +## Telegram migration producer + +Telegram group migration reactivates conversations before the generic resource +availability helper runs. A real owner regression at b634353059 fails to publish +the existing failed run after commit (/tmp/sleep28-resource-telegram-before.log). +The Telegram module now locks the bindings it moves and registers distinct issue +milestone intents before its existing update. It preserves the caller's outer +transaction, current migration/authorization semantics, and canonical topic IDs. +No shared scheduler contract or new durable table is introduced. + +All four resource producer cases pass, including nested-savepoint/outer rollback, +commit reactivation, unrelated completed history, two minutes of quiet, unchanged +inventory and duplicate Telegram migration without a new milestone intent: +/tmp/sleep28-resource-telegram-tests.log. All 88 existing lifecycle/governed cases +pass in /tmp/sleep28-resource-telegram-lifecycle.log. Eight existing Telegram +migration integration cases pass in /tmp/sleep28-resource-telegram-acceptance.log +(1,422 deselected): unset/inherited/cleared/configured execution defaults, with +and without an existing destination, topic identity and subsequent reply routing. +Server production typecheck and build pass in the corresponding +/tmp/sleep28-resource-telegram-{typecheck,build}.log files. Total: 100 distinct +selected tests; no live Telegram or host-suspension qualification is claimed. + +The cancellation E2E diagnostic completed at b634353059: focused two cases pass, +the full Agent Chat projects file passes all 16, and the two cases with a +100ms database-only running-row delay pass. Evidence is retained in +/tmp/sleep-cancel-e2e-20261025, /tmp/sleep-cancel-full-e2e-20261025 and +/tmp/sleep-cancel-delayed-e2e-20261025. Captured lock waits do not establish the +CI deadlock cycle. These passing runs do not resolve the actual #17572/#17490 +E2E failures; no blind retry or cancellation-error masking was applied. + +Normal inbound conversation/link activation and remaining authority/lineage +producer dispositions still gate production registration and polling removal. +Fresh master is 7ebf98eafd; final base reconciliation remains outstanding. diff --git a/server/src/__tests__/chat-run-milestone-resource-work.test.ts b/server/src/__tests__/chat-run-milestone-resource-work.test.ts index c760d6168c7..8574ba2a70a 100644 --- a/server/src/__tests__/chat-run-milestone-resource-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-resource-work.test.ts @@ -49,10 +49,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } const publications = () => db.select().from(chatPublications).where(eq(chatPublications.issueId, f.issueId)); - it.each(["inventory", "lifecycle", "github repository"] as const)("wakes only reopened conversations after %s commits, preserving rollback and completed history", async path => { - const provider = path === "github repository" ? "github" : "slack"; - const resourceType = provider === "github" ? "repository" : "channel"; - const providerResourceId = provider === "github" ? "owner/repo" : "recovered-channel"; + it.each(["inventory", "lifecycle", "github repository", "telegram migration"] as const)("wakes only reopened conversations after %s commits, preserving rollback and completed history", async path => { + const provider = path === "github repository" ? "github" : path === "telegram migration" ? "telegram" : "slack"; + const resourceType = provider === "github" ? "repository" : provider === "telegram" ? "chat" : "channel"; + const providerResourceId = provider === "github" ? "owner/repo" : provider === "telegram" ? "-100222" : "recovered-channel"; await db.update(chatEndpoints).set({ provider }).where(eq(chatEndpoints.id, f.endpointId)); const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, @@ -62,12 +62,14 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, f.issueId)); const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId)); const [resource] = await db.insert(chatEndpointResources).values({ companyId: f.companyId, endpointId: f.endpointId, - type: resourceType, providerResourceId, label: "Recovered", availability: "unavailable", enabled: true, + type: resourceType, providerResourceId: provider === "telegram" ? "-111" : providerResourceId, label: "Recovered", availability: "unavailable", enabled: true, }).returning(); - await db.update(chatConversations).set({ resourceId: resource.id, state: "unavailable" }).where(eq(chatConversations.id, f.conversationId)); + await db.update(chatConversations).set({ resourceId: resource.id, state: "unavailable", ...(provider === "telegram" ? { externalConversationId: "telegram:-111", externalThreadId: "telegram:-111:42" } : {}) }).where(eq(chatConversations.id, f.conversationId)); const completedId = randomUUID(); await db.insert(chatConversations).values({ id: completedId, companyId: f.companyId, endpointId: f.endpointId, - resourceId: resource.id, issueId: f.issueId, state: "completed", externalConversationId: completedId, externalThreadId: completedId, externalLabel: "History" }); + // Telegram migration intentionally rebinds all conversations on the old resource. + // Unrelated completed history must remain passive. + resourceId: provider === "telegram" ? null : resource.id, issueId: f.issueId, state: "completed", externalConversationId: completedId, externalThreadId: completedId, externalLabel: "History" }); const [delivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: f.endpointId, providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "installation", normalizedEvent: {}, state: "processing", attempts: 1, }).returning(); @@ -75,7 +77,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await chatRunMilestoneWork(db).register(coordinator, () => true).ready; await flush(); expect(await publications()).toEqual([]); await quiet(); const withLease = chatCredentialMutationLease(db); - const recover = (rollback: boolean) => withLease(endpoint, lease => db.transaction(async tx => { + const recover = (rollback: boolean, alreadyPublished = false) => withLease(endpoint, lease => db.transaction(async tx => { const resources = createChatResourceReconciliation(tx as unknown as typeof db, { endpointRecord: async () => ({ endpoint }), withCredentialMutationLease: withLease, runtimeGeneration: () => 0, maxErrorText: 1000, }); @@ -85,12 +87,13 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await resources.reconcileGitHubWebhookRepository(endpoint, { repository: { id: 123, full_name: providerResourceId, name: "repo", owner: { id: 321, login: "owner" } } }, lease); } else { await resources.applyProviderAvailability({ currentEndpoint: endpoint, credentialLease: lease, staleResourceEffect: false, - claimed: delivery, candidate: { attempts: 0 }, effect: { kind: "resource", provider: "slack", providerEventId: delivery.providerEventId, - providerResourceId: "recovered-channel", resourceType: "channel", label: "Recovered", availability: "available" }, + claimed: delivery, candidate: { attempts: 0 }, effect: { kind: "resource", provider, providerEventId: delivery.providerEventId, + providerResourceId, resourceType, ...(provider === "telegram" ? { previousProviderResourceId: "-111" } : {}), label: "Recovered", availability: "available" }, }); } // The owner used a nested transaction. Its savepoint release cannot publish. - await vi.advanceTimersByTimeAsync(1000); expect(await publications()).toEqual([]); + await vi.advanceTimersByTimeAsync(1000); + expect(await publications()).toHaveLength(alreadyPublished ? 1 : 0); if (rollback) throw new Error("outer rollback"); })); await expect(recover(true)).rejects.toThrow("outer rollback"); @@ -98,6 +101,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await recover(false); await flush(); expect(await publications()).toEqual([expect.objectContaining({ idempotencyKey: `run:${runId}:failed:${f.endpointId}` })]); expect((await db.select().from(chatConversations).where(eq(chatConversations.id, completedId)))[0].state).toBe("completed"); + if (provider === "telegram") { + expect((await db.select().from(chatConversations).where(eq(chatConversations.id, f.conversationId)))[0]) + .toMatchObject({ externalConversationId: "telegram:-100222", externalThreadId: "telegram:-100222:42", state: "active" }); + } await quiet(); // Unchanged inventory must not turn a provider refresh into milestone polling. const notices: string[] = []; @@ -107,6 +114,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu endpointRecord: async () => ({ endpoint }), withCredentialMutationLease: withLease, runtimeGeneration: () => 0, maxErrorText: 1000, }).reconcileProviderResourceRows(endpoint, { provider, resources: [{ type: resourceType, providerResourceId, label: "Recovered" }] }, lease)); expect(notices).toEqual([]); + if (provider === "telegram") { + // A duplicate migration has no old bindings left to reactivate. + await recover(false, true); + expect(notices).toEqual([]); + } } finally { unsubscribe(); } await flush(); await quiet(); }); diff --git a/server/src/services/chat-channels/telegram/resources.ts b/server/src/services/chat-channels/telegram/resources.ts index 17bc3b39b52..1e8174afe78 100644 --- a/server/src/services/chat-channels/telegram/resources.ts +++ b/server/src/services/chat-channels/telegram/resources.ts @@ -1,5 +1,6 @@ import { and, eq, sql } from "drizzle-orm"; import { chatConversations, chatEndpointResources, chatEndpoints, type Db } from "@paperclipai/db"; +import { notifyChatRunMilestones } from "../../chat-run-milestone-notifications.js"; import type { ChatProviderLifecycleEffect } from "../../chat-provider-lifecycle.js"; type DbTransaction = Parameters[0]>[0]; type EndpointRow = typeof chatEndpoints.$inferSelect; @@ -70,6 +71,18 @@ export async function migrateTelegramResourceConversations( resource: { id: string }, previousProviderResourceId: string | undefined, providerResourceId: string, migratedLabel: string, now: Date, ) { + const affected = and( + eq(chatConversations.companyId, currentEndpoint.companyId), + eq(chatConversations.endpointId, currentEndpoint.id), + eq(chatConversations.resourceId, previousResource.id), + ); + const migrated = await tx.select({ issueId: chatConversations.issueId }) + .from(chatConversations).where(affected).for("update"); + // Migration restores the binding before generic resource availability runs. + // Register the wake here, within the same owning transaction as that change. + for (const issueId of new Set(migrated.map(row => row.issueId))) { + await notifyChatRunMilestones(tx, { companyId: currentEndpoint.companyId, issueId }); + } const previousThreadPrefix = `telegram:${previousProviderResourceId}`; const migratedThreadPrefix = `telegram:${providerResourceId}`; await tx @@ -94,11 +107,5 @@ export async function migrateTelegramResourceConversations( state: "active", updatedAt: now, }) - .where( - and( - eq(chatConversations.companyId, currentEndpoint.companyId), - eq(chatConversations.endpointId, currentEndpoint.id), - eq(chatConversations.resourceId, previousResource.id), - ), - ); + .where(affected); }