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 8a360ef8d07..a54d9b1898b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1586,3 +1586,33 @@ their owning branches, and propagate. Their currently passing CI is not evidence against that missing idle-counter assertion. Then continue milestone configuration and alternate-receipt coverage, production registration/poll removal, the entire original inventory, older E2E/cleanup findings and final master reconciliation. + +## Configuration drain propagation and resource milestone producers + +Discord #17557 at 954a0faa75 and receiver #17572 at 8815dfad38 now carry the +confirmed drain-hold repairs, following tool health #17554 at e685a0e7ef. +The prerequisite #17561 inherited health at 05ac7835fc. Propagation reached +#17576 a4bc3ea50e, #17578 05797e2b9f, #17580 82d023a3a0, #17583 f4f7e14f7a, +#17586 920a951505, #17590 d626ff4b13, #17592 33ffff8a12, and #17606 9775b2bc4a. +All 109 configuration lifecycle/protocol cases pass together after applying the +tip schema to their disposable databases. See the configuration-worker drain +repair document for individual causal failures and parent evidence. + +Slice 28 adds real inventory/lifecycle resource reactivation wakes. The two +regressions fail before repair; all 90 milestone lifecycle/governed/resource cases +and six selected provider acceptance cases pass after it. The owning resource +module only wakes reopened issue scopes, never unchanged inventory or completed +history. See the milestone audit for full commands/results and remaining producers. + +Fresh gate snapshot /tmp/sleep-stack-gates-configuration-drain-20261025.json has +no failures for slices 1–12 and 14–16, plus #17505; #17490 still fails E2E shard 3 +and its aggregate. #17518/#17554 have pending review and later updated heads have +pending checks. New #17576 security finding 4371283330 concerns repeated +connection writes loading every live voice session in that connection. Inspect +and address it without dropping future deadlines or configuration revocation. +The existing #17580 cleanup/company-deletion denial finding remains open. No +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. 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 d083d928f32..734869b10b1 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 @@ -1072,3 +1072,58 @@ passing evidence; the full 108 voice integration cases passed at its parent. This establishes the changed dispatch/drain boundary, not final PR readiness or host sleep. The same pending predicate in three earlier independent workers is still a required parent repair. Remaining milestone producers/wiring are unchanged. + +## Resource reactivation producers (2026-10-25) + +The current selector admits active/waiting conversations. Resource inventory and +provider lifecycle recovery can reopen an unavailable conversation after the +milestone worker exhausted its earlier scan. Both paths lacked an issue wake. +The real-owner regressions in /tmp/sleep28-resource-before.log fail after commit +with no failed-run publication, establishing the gap independently of manual +notification helpers. + +Resource reconciliation now owns one local availability transition function. It +locks the unavailable bindings it will reopen, signals their distinct issue +scopes before mutation, and retains the existing availability/completed-history +rules. Both inventory and lifecycle callers use it. Unchanged inventory produces +no milestone intent; status/health metadata is not indiscriminately promoted to +company-wide milestone scans. No shared scheduler contract or provider protocol +changes. The tests use the real credential lease and nested resource transaction, +prove savepoint release does not publish, prove outer rollback leaves the worker +quiet, and prove outer commit creates the expected publication. Completed history +stays completed. They then verify two minutes of quiet and no intent from a +repeated unchanged inventory. + +All 90 milestone lifecycle/governed/resource cases pass in +/tmp/sleep28-resource-combined-tests.log. That command also named a nonexistent +__tests__/chat-run-publications.test.ts path; it did not run projection units. +The separate corrected service-path run is recorded in +/tmp/sleep28-resource-projection-tests.log. Six existing provider acceptance cases +pass in /tmp/sleep28-resource-acceptance.log (1,424 deselected, not counted), covering +Slack membership/reinstall, Slack archive/unarchive, Teams/Telegram lifecycle, +GitHub resource identity preservation, durable lifecycle retry, and canonical +Chat SDK resource identifiers. Server typecheck passes in +/tmp/sleep28-resource-typecheck.log. The dedicated resource database is +paperclip_sleep28_resource_20261025_test. + +The earlier three configuration-owner drain repairs are now propagated. All 109 +health/Discord/receiver cases pass together at 9775b2bc4a in +/tmp/sleep28-configuration-drain-tests.log. An earlier combined run had 108 passes +and one schema mismatch because the parent test databases lacked the downstream +voice cleanup column; the retained log is +/tmp/sleep28-configuration-drain-schema-before.log. Applying the actual tip +migrations to those disposable databases fixes that setup error. + +This closes resource-owner reactivation, not every configuration producer. +GitHub-specific repository/installation restoration, other conversation rebinding, +late authority/lineage recovery, and remaining producer dispositions still need +review before production registration and retirement of the lifecycle scan. +Endpoint status/health are not direct milestone selector predicates; changing +them alone does not justify wiring a broad company wake to every health check. +The delivery owner retains its current authority checks. + +Final resource checkpoint: all 20 projection units pass at the corrected path in +/tmp/sleep28-resource-projection-tests.log, and server build passes in +/tmp/sleep28-resource-build.log. Combined selected evidence is 116 distinct +milestone/projection/provider cases, plus the 109 propagated configuration-owner +cases. Full remote checks/reviews remain required for the new head. diff --git a/server/src/__tests__/chat-run-milestone-resource-work.test.ts b/server/src/__tests__/chat-run-milestone-resource-work.test.ts new file mode 100644 index 00000000000..2236f5952bd --- /dev/null +++ b/server/src/__tests__/chat-run-milestone-resource-work.test.ts @@ -0,0 +1,107 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { chatConversations, chatDeliveries, chatEndpointResources, chatEndpoints, chatPublications, createDb, heartbeatRuns, issues, subscribeDatabaseWork } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createChatMilestoneFixtures } from "./helpers/chat-milestone-fixtures.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { chatRunMilestoneWork } from "../services/chat-run-milestone-work.js"; +import { chatCredentialMutationLease } from "../services/chat-credential-mutation-lease.js"; +import { createChatResourceReconciliation } from "../services/chat-channels/resource-reconciliation.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MILESTONE_RESOURCE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("chat milestones after resource recovery", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + let fixtures: ReturnType; + let f: Awaited>; + let coordinator: ReturnType | undefined; + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-resource-milestones-"); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + fixtures = createChatMilestoneFixtures(db); + }, 60_000); + beforeEach(async () => { + errors.length = 0; f = await fixtures.create(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(chatDeliveries).where(eq(chatDeliveries.companyId, f.companyId)); + await fixtures.cleanup(); expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + } + async function quiet() { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + const publications = () => db.select().from(chatPublications).where(eq(chatPublications.issueId, f.issueId)); + + it.each(["inventory", "lifecycle"] as const)("wakes only reopened conversations after %s commits, preserving rollback and completed history", async path => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, + responsibleUserId: f.userId, status: "failed", errorCode: "test_run_failed", updatedAt: new Date(), + contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId }, + }); + 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: "channel", providerResourceId: "recovered-channel", label: "Recovered", availability: "unavailable", enabled: true, + }).returning(); + await db.update(chatConversations).set({ resourceId: resource.id, state: "unavailable" }).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" }); + 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(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + 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 resources = createChatResourceReconciliation(tx as unknown as typeof db, { + endpointRecord: async () => ({ endpoint }), withCredentialMutationLease: withLease, runtimeGeneration: () => 0, maxErrorText: 1000, + }); + if (path === "inventory") { + await resources.reconcileProviderResourceRows(endpoint, { provider: "slack", resources: [{ type: "channel", providerResourceId: "recovered-channel", label: "Recovered" }] }, 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" }, + }); + } + // The owner used a nested transaction. Its savepoint release cannot publish. + await vi.advanceTimersByTimeAsync(1000); expect(await publications()).toEqual([]); + if (rollback) throw new Error("outer rollback"); + })); + await expect(recover(true)).rejects.toThrow("outer rollback"); + await flush(); expect(await publications()).toEqual([]); await quiet(); + 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"); + await quiet(); + // Unchanged inventory must not turn a provider refresh into milestone polling. + const notices: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => notices.push(event)); + try { + await withLease(endpoint, lease => createChatResourceReconciliation(db, { + endpointRecord: async () => ({ endpoint }), withCredentialMutationLease: withLease, runtimeGeneration: () => 0, maxErrorText: 1000, + }).reconcileProviderResourceRows(endpoint, { provider: "slack", resources: [{ type: "channel", providerResourceId: "recovered-channel", label: "Recovered" }] }, lease)); + expect(notices).toEqual([]); + } finally { unsubscribe(); } + await flush(); await quiet(); + }); +}); diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 0b52e8e7c69..17eb86ec6d9 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; @@ -79,6 +80,32 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc return resource ?? null; } + async function setResourceConversationAvailability( + tx: DbTransaction, + endpoint: Pick, + resourceId: string, + available: boolean, + now: Date, + ) { + const affected = and( + eq(chatConversations.companyId, endpoint.companyId), + eq(chatConversations.endpointId, endpoint.id), + eq(chatConversations.resourceId, resourceId), + available ? eq(chatConversations.state, "unavailable") : inArray(chatConversations.state, ["active", "waiting"]), + ); + if (available) { + // Only a reopened binding creates projection work. Repeated inventory + // refreshes and completed history must not wake the milestone scanner. + const reopened = await tx.select({ issueId: chatConversations.issueId }) + .from(chatConversations).where(affected).for("update"); + for (const issueId of new Set(reopened.map(row => row.issueId))) { + await notifyChatRunMilestones(tx, { companyId: endpoint.companyId, issueId }); + } + } + await tx.update(chatConversations) + .set({ state: available ? "active" : "unavailable", updatedAt: now }).where(affected); + } + async function reconcileProviderResourceRows( endpoint: EndpointRow, inventory: ChatProviderInventoryResult, @@ -98,17 +125,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc if (resource) { // Resource recovery reopens only bindings that Paperclip marked // unavailable. Completed historical tasks stay completed. - await tx - .update(chatConversations) - .set({ state: "active", updatedAt: new Date() }) - .where( - and( - eq(chatConversations.companyId, endpoint.companyId), - eq(chatConversations.endpointId, endpoint.id), - eq(chatConversations.resourceId, resource.id), - eq(chatConversations.state, "unavailable"), - ), - ); + await setResourceConversationAvailability(tx, endpoint, resource.id, true, new Date()); } } @@ -131,17 +148,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .update(chatEndpointResources) .set({ availability: absentAvailability, updatedAt: new Date() }) .where(eq(chatEndpointResources.id, resource.id)); - await tx - .update(chatConversations) - .set({ state: "unavailable", updatedAt: new Date() }) - .where( - and( - eq(chatConversations.companyId, endpoint.companyId), - eq(chatConversations.endpointId, endpoint.id), - eq(chatConversations.resourceId, resource.id), - inArray(chatConversations.state, ["active", "waiting"]), - ), - ); + await setResourceConversationAvailability(tx, endpoint, resource.id, false, new Date()); } await credentialLease.assertOwned(tx); }); @@ -470,25 +477,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc await migrateTelegramResourceConversations(tx, currentEndpoint, previousResource, resource, previousProviderResourceId, providerResourceId, migratedLabel, now); } if (resource) { - await tx - .update(chatConversations) - .set({ - state: - effect.availability === "available" - ? "active" - : "unavailable", - updatedAt: now, - }) - .where( - and( - eq(chatConversations.companyId, currentEndpoint.companyId), - eq(chatConversations.endpointId, currentEndpoint.id), - eq(chatConversations.resourceId, resource.id), - effect.availability === "available" - ? eq(chatConversations.state, "unavailable") - : inArray(chatConversations.state, ["active", "waiting"]), - ), - ); + await setResourceConversationAvailability(tx, currentEndpoint, resource.id, effect.availability === "available", now); } await tx .update(chatEndpoints)