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 1ea42d00856..674fe474361 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1651,3 +1651,12 @@ audit retains the initial contention-sensitive assertion and its correction to honor the existing retry. Identity activation and remaining authority/lineage producers still gate production polling removal. All original slices and older parent CI/review findings remain required. + + +Slice 28 identity activation now signals the bound issue before changing the +responsible user. Two before-fix failures now pass with outer commit/rollback, +pending steering, replay and settled quiet checks. All 43 existing identity and +GitHub credential tests pass after repairing their external-DB fixture URL; +server types/build pass. Recovery terminalization/classification writers are the +next audit. Master 34409a92c4 was fetched without disturbing this draft. The +worker is still not production-registered and all original stack duties remain. 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 d0a0e09432d..9bd78bf6f06 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 @@ -1264,3 +1264,29 @@ strategy-only wake changes do not alter lineage fields. Slack Board dispatch use source issue.comment and its own response path; do not infer milestone eligibility from the shared binding reader alone. These are dispositions and next audits, not proof that all remaining producers are complete. + + +## Run identity activation + +Monitor/continuation lineage requires matching responsibleUserId on source and +successor. Initialization and accepted steering can restore that match after the +worker exhausted its candidates. The identity owner now reads the bound issue +and registers its intent before the existing activation write. Pending steering +and replayed acceptance do not activate work. Production callers already own a +transaction; standalone tests now use reconcileSteeredIdentity, the existing +transaction/lock owner, instead of accepting directly on a raw client. + +Two real-PostgreSQL worker cases reproduced missing notices before the fix and +pass afterward: initialization and accepted steering, each with pending/quiet +state, outer rollback, outer commit, replay, and settled two-minute SQL quiet. +Evidence: /tmp/sleep28-identity-before.log and /tmp/sleep28-identity-tests.log. +All 43 existing identity/GitHub credential tests pass in +/tmp/sleep28-identity-acceptance-fixed.log. The first acceptance run found two +lock fixtures still using the embedded-only URL; those now use the configured +connection URL, retain their one-second lock timeout, and close the extra pool. +Server typecheck/build passed in /tmp/sleep28-identity-{typecheck,build}.log. + +Fresh master is 34409a92c4 (native failover backup filtering); it was fetched, +not merged into this draft. Recovery service orphan terminalization and adapter +failure classification are the next direct run writers to audit. Production +registration/poll removal and the original stack's gates remain unfinished. diff --git a/server/src/__tests__/chat-run-milestone-identity-work.test.ts b/server/src/__tests__/chat-run-milestone-identity-work.test.ts new file mode 100644 index 00000000000..76d3a0376fc --- /dev/null +++ b/server/src/__tests__/chat-run-milestone-identity-work.test.ts @@ -0,0 +1,104 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agentWakeupRequests, chatPublications, createDb, heartbeatRuns, issueComments, 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 { initializeRunIdentity, reserveSteeredIdentity, acceptSteeredIdentity } from "../services/run-identity.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MILESTONE_IDENTITY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("chat milestones after identity activation", () => { + 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-identity-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 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(["initialization", "steering"] as const)("wakes matching monitor lineage after %s commits, preserving rollback and replay", async kind => { + const nextCheckAt = new Date(Date.now() - 1000).toISOString(); + const sourceId = randomUUID(), successorId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: sourceId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "succeeded", responsibleUserId: f.userId, + contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId }, + resultJson: { semanticToolReceipts: { schedule: { operationId: "set_task_monitor", + result: { taskId: f.issueId, monitor: { status: "scheduled", nextCheckAt } } } } }, + }); + await db.insert(heartbeatRuns).values({ id: successorId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "running", responsibleUserId: null, + contextSnapshot: { issueId: f.issueId, source: "issue.monitor" }, + }); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, runId: successorId, + source: "automation", reason: "issue_monitor_due", status: "claimed", + idempotencyKey: `issue-monitor:${f.issueId}:${nextCheckAt}`, payload: { issueId: f.issueId, nextCheckAt }, + }); + const identityInput = { companyId: f.companyId, runId: successorId, issueId: f.issueId, responsibleUserId: f.userId, cause: "dispatch" }; + if (kind === "steering") { + await initializeRunIdentity(db, { ...identityInput, responsibleUserId: null }); + await db.update(issueComments).set({ authorUserId: f.userId }).where(eq(issueComments.id, f.commentId)); + } + 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 pending = kind === "steering" ? await reserveSteeredIdentity(db, { + companyId: f.companyId, runId: successorId, issueId: f.issueId, messageId: f.commentId, + }) : null; + // Reserving an unacknowledged message must not activate its identity. + if (kind === "steering") expect(pending?.status).toBe("pending"); + await flush(); expect(await publications()).toEqual([]); await quiet(); + const activate = (rollback: boolean) => db.transaction(async tx => { + if (kind === "steering") await acceptSteeredIdentity(tx, pending!); + else await initializeRunIdentity(tx as unknown as typeof db, identityInput); + await vi.advanceTimersByTimeAsync(1000); + expect(await publications()).toEqual([]); + if (rollback) throw new Error("outer rollback"); + }); + await expect(activate(true)).rejects.toThrow("outer rollback"); + await flush(); expect(await publications()).toEqual([]); await quiet(); + await activate(false); await flush(); + expect(await publications()).toEqual([expect.objectContaining({ idempotencyKey: `run:${successorId}:working:${f.endpointId}` })]); + await quiet(); + const notices: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => notices.push(event)); + try { + await db.transaction(async tx => { + if (kind === "steering") await acceptSteeredIdentity(tx, pending!); + else await initializeRunIdentity(tx as unknown as typeof db, identityInput); + }); + expect(notices).toEqual([]); + } finally { unsubscribe(); } + await flush(); await quiet(); + }); +}); diff --git a/server/src/__tests__/github-operation-credentials.test.ts b/server/src/__tests__/github-operation-credentials.test.ts index 02d1c864bcd..b072798daac 100644 --- a/server/src/__tests__/github-operation-credentials.test.ts +++ b/server/src/__tests__/github-operation-credentials.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createNativeGitHubAccess } from "../services/native-runtime/native-github-access.js"; import express from "express"; import request from "supertest"; @@ -32,7 +33,7 @@ import { import { initializeRunIdentity, reserveSteeredIdentity, - acceptSteeredIdentity, + reconcileSteeredIdentity, } from "../services/run-identity.js"; import { resolveGitHubOperationCredentials } from "../services/github-operation-credentials.js"; import { @@ -49,23 +50,25 @@ const vault = vi.hoisted(() => ({ resolveSecretValue: vi.fn(async () => "test-dedicated-token"), })); vi.mock("../services/secrets.js", () => ({ secretService: () => vault })); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_GITHUB_CREDENTIAL_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)( "operation-time GitHub credential resolution", () => { - let database: Awaited>, + let database: Awaited> | undefined, db: ReturnType; beforeAll(async () => { vi.stubEnv( "PAPERCLIP_AGENT_JWT_SECRET", "test-github-broker-signing-secret", ); - database = await startEmbeddedPostgresTestDatabase( + if (!external) database = await startEmbeddedPostgresTestDatabase( "paperclip-github-operation-", ); - db = createDb(database.connectionString); + db = createDb(external ?? database!.connectionString); }, 30_000); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); vi.unstubAllEnvs(); }, 60_000); @@ -208,7 +211,7 @@ const support = await getEmbeddedPostgresTestSupport(); ...input, messageId: id, }); - await acceptSteeredIdentity(db, context!); + await reconcileSteeredIdentity(db, context!); } it("resolves A → B → A without retaining tokens, and records only redacted diagnostics", async () => { const input = await seed(); diff --git a/server/src/__tests__/run-identity.test.ts b/server/src/__tests__/run-identity.test.ts index d6dbbbfc3ff..15aca0c9eaf 100644 --- a/server/src/__tests__/run-identity.test.ts +++ b/server/src/__tests__/run-identity.test.ts @@ -1,19 +1,21 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, heartbeatRunEvents, issueComments, issueThreadInteractions, issues, secretAccessEvents } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -import { acceptSteeredIdentity, captureRunIdentity, initializeRunIdentity, listRunIdentityContexts, rejectSteeredIdentity, reserveSteeredIdentity } from "../services/run-identity.js"; +import { acceptSteeredIdentity, reconcileSteeredIdentity, captureRunIdentity, initializeRunIdentity, listRunIdentityContexts, rejectSteeredIdentity, reserveSteeredIdentity } from "../services/run-identity.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_IDENTITY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("durable execution identity", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: ReturnType; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-run-identity-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-run-identity-"); + db = createDb(external ?? database!.connectionString); }, 30_000); - afterAll(async () => { await database?.cleanup(); }, 60_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }, 60_000); async function seed() { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); @@ -121,8 +123,8 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("B"); expect(startedUnderA.context?.responsibleUserId).toBe("A"); const a = await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[2] }); - await acceptSteeredIdentity(db, a!); - await acceptSteeredIdentity(db, b!); + await reconcileSteeredIdentity(db, a!); + await reconcileSteeredIdentity(db, b!); expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("A"); }); it("keeps rejected steering unchanged and preserves the originating identity on retry", async () => { @@ -180,7 +182,7 @@ const support = await getEmbeddedPostgresTestSupport(); sourceIdentityContextId: origin.id, payload: { prompt: "Continue?" } as never, }); const b = await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[1] }); - await acceptSteeredIdentity(db, b!); + await reconcileSteeredIdentity(db, b!); for (const approval of [true, false]) { const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, companyId: input.companyId, agentId: input.agentId, status: "running" }); @@ -251,34 +253,40 @@ const support = await getEmbeddedPostgresTestSupport(); // A real append holds KEY SHARE on both parent rows until it commits. // Bound the other connection's wait so a conflicting lock fails this // regression instead of leaving both transactions waiting for each other. - const identityDb = createDb(`${database.connectionString}?options=-c%20lock_timeout%3D1000`, { + const identityUrl = new URL(external ?? database!.connectionString); + identityUrl.searchParams.set("options", "-c lock_timeout=1000"); + const identityDb = createDb(identityUrl.toString(), { maxConnections: 1, }); - const [settings] = await identityDb.execute(sql`show lock_timeout`); - expect(settings?.lock_timeout).toBe("1s"); - await db.transaction(async (audit) => { - await audit.insert(secretAccessEvents).values({ - companyId: input.companyId, - heartbeatRunId: input.runId, - issueId: input.issueId, - provider: "local_encrypted", - actorType: "agent", - actorId: input.agentId, - consumerType: "agent", - consumerId: input.agentId, - outcome: "granted", - }); - if (operation === "initialize") { - await expect(initializeRunIdentity(identityDb, { - ...input, messageIds: [], responsibleUserId: "A", cause: "instruction", - })).resolves.toMatchObject({ responsibleUserId: "A" }); - } else { - await expect(captureRunIdentity(identityDb, input)).resolves.toMatchObject({ - run: { responsibleUserId: "A" }, - context: { responsibleUserId: "A" }, + try { + const [settings] = await identityDb.execute(sql`show lock_timeout`); + expect(settings?.lock_timeout).toBe("1s"); + await db.transaction(async (audit) => { + await audit.insert(secretAccessEvents).values({ + companyId: input.companyId, + heartbeatRunId: input.runId, + issueId: input.issueId, + provider: "local_encrypted", + actorType: "agent", + actorId: input.agentId, + consumerType: "agent", + consumerId: input.agentId, + outcome: "granted", }); - } - }); + if (operation === "initialize") { + await expect(initializeRunIdentity(identityDb, { + ...input, messageIds: [], responsibleUserId: "A", cause: "instruction", + })).resolves.toMatchObject({ responsibleUserId: "A" }); + } else { + await expect(captureRunIdentity(identityDb, input)).resolves.toMatchObject({ + run: { responsibleUserId: "A" }, + context: { responsibleUserId: "A" }, + }); + } + }); + } finally { + await identityDb.$client.end({ timeout: 1 }); + } }, ); diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 19ac75989ce..18541161c96 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -10,7 +10,8 @@ import { type Db, } from "@paperclipai/db"; import { conflict, forbidden } from "../errors.js"; -import { boundRunIssueId } from "./run-task-binding.js"; +import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; +import { boundRunIssueId, boundRunIssueIdSql } from "./run-task-binding.js"; import { isUuidLike } from "@paperclipai/shared"; import { queuedCommentIdsFromRunContext, queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; @@ -137,6 +138,13 @@ async function append( } async function activate(executor: Executor, context: RunIdentityContext) { + const [binding] = await executor.select({ issueId: boundRunIssueIdSql() }) + .from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, context.runId), eq(heartbeatRuns.companyId, context.companyId), + )); + if (binding?.issueId) { + await notifyChatRunMilestones(executor, { companyId: context.companyId, issueId: binding.issueId }); + } const [run] = await executor .update(heartbeatRuns) .set({ @@ -409,6 +417,7 @@ export async function rejectSteeredIdentity( ); } +/** Accept within the caller's task/run transaction; standalone recovery uses reconcileSteeredIdentity. */ export async function acceptSteeredIdentity( executor: Executor, context: RunIdentityContext,