fix(server): close milestone identity and recovery wake gaps

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 11:52:03 -05:00
1 parent 1b0bec0b25
commit ed76256b97
8 files changed
+202 -64

No files matched your search

@@ -1710,3 +1710,12 @@ audit has a consolidated alternate-writer disposition table. Native progress
remains the next separate reason. Remote gates, parent findings and the rest of
the original stack are not complete; local cancellation success does not repair
the recorded CI deadlock.
Slice 28 review repair: cover GitHub observed-identity linking, suppress
unchanged native-finalization recovery wakes, and repair the voice callback and
Cloud-auth test contracts. Restore shared user profile after the Cloud authority
test so downstream Slack publication tests see their original identity. Local
40 stale-lock, 122 voice, 38 auth/session and 28 combined chat cases pass; server
typecheck/build pass. See the milestone audit for failed-run dispositions.
Latest-head remote gates and parent failures remain open.
@@ -1522,3 +1522,58 @@ for the unresolved cancellation investigation, not proof of its entire lock cycl
No blind rerun or masking was performed. Production milestone registration,
final producer dispositions, native progress as its separate slice, remote gates,
and every remaining original stack duty remain outstanding.
## Latest review and CI repair (2026-10-25)
Review of `1b0bec0b25` found three defects. All were reproduced before repair:
- GitHub `linkObservedIdentity` is a separate supported identity writer. It now
signals before creating/reactivating its identity link, inside the existing
membership and principal locks. Reconfirming an already-linked identity is
silent. The real producer test covers outer commit, rollback and reconfirmation.
- Orphan-run terminalization registered intent even when its guarded update
rejected retained native finalization. It now locks an eligible run before
intent and preserves the same predicate on the update. Five retained phases,
each swept three times, emit no milestone signals. All 40 stale-lock tests pass,
including coordinator-arrival races, terminal-issue authority and exhausted
finalization recovery. This does not migrate the stale-lock polling owner.
- The voice retention integration test still supplied the former coordinator
callback name. It now uses `projectNativeProgress`; the original failure was
`input.projectNativeProgress is not a function`.
CI run `40212695778` also exposed an incomplete Cloud-auth fake and shared-user
fixture leakage. The fake now models transaction and locking calls, retaining
failure injection for optional membership/settings reads. The Cloud authority
integration test restores the shared user's name/email/verification metadata in
its finally block. Previously it left `Owner 1` behind, breaking later Slack
publication assertions that expected `Owner User`. No publication policy or
assertion was weakened to hide the changed name.
Evidence:
- `/tmp/sleep28-review-before.log`: five redundant-wake regressions and the
missing GitHub identity intent fail before repair.
- `/tmp/sleep28-voice-callback-before.log`: reproduces the renamed-callback error.
- `/tmp/sleep28-review-fixed.log`: all 40 stale-lock tests pass; one unscoped
signal assertion observed another fixture's notification. The two GitHub
transaction assertions now filter the actual company scope; both pass in
`/tmp/sleep28-review-github-scoped.log`.
- `/tmp/sleep28-voice-callback-fixed.log`: 28 failures on a reused database with
retained voice sessions/reports. Preserve this failed run. The full suite on
fresh migrated `paperclip_sleep28_review_voice_20261025_test` passes all 122
cases (`/tmp/sleep28-review-voice-fresh.log`). Global reconstruction tests need
an isolated database; reusing a previous suite's retained obligations is invalid.
- `/tmp/sleep28-review-cloud-fakes-final.log`: 38 Cloud-auth/session tests pass.
The first repair exposed an obsolete one-read assertion; the updated assertion
still verifies the exact own-user/active filter on the authorization read.
- `/tmp/sleep28-review-long-board.log`: reproduces the `Owner 1` fixture leak.
- `/tmp/sleep28-review-chat-isolation.log`: 28 cases pass together on fresh
`paperclip_sleep28_review_chat_20261025_test`: all ten authority-restoration
cases, both GitHub signal cases, and all 16 long-publication cases.
- `/tmp/sleep28-review-typecheck.log` and `/tmp/sleep28-review-build.log`: server
production typecheck and build pass. Test files are not typechecked.
Latest-head CI/review and stack reconciliation remain required. These local
passes do not discharge previous cancellation/deadlock evidence or parent review
findings. No merge or host-sleep qualification is claimed.
@@ -5542,6 +5542,41 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
.where(eq(chatEndpointLeases.endpointId, f.endpoint.id)),
).toHaveLength(0);
});
it("wakes GitHub milestone identity restoration only after outer commit and stays quiet on reconfirmation", async () => {
const f = await reviewBotFixture();
await f.service.revokeLink(f.endpoint.id, f.principal.id);
const verified = { githubUserId: "42", login: "octocat" };
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, (event, scopes) => {
if ([...scopes ?? []].some(scope => JSON.parse(scope).companyId === f.companyId)) events.push(event);
});
const link = () => db.select().from(chatIdentityLinks).where(and(
eq(chatIdentityLinks.endpointId, f.endpoint.id), eq(chatIdentityLinks.principalId, f.principal.id),
)).then(rows => rows[0]);
try {
await expect(db.transaction(async tx => {
await githubChatManagementService(tx as unknown as typeof db, f.providerFetch)
.linkObservedIdentity(f.endpoint.id, "owner-user", verified);
expect(events).toEqual(["intent"]);
expect((await link()).status).toBe("revoked");
throw new Error("rollback identity restoration");
})).rejects.toThrow("rollback identity restoration");
expect((await link()).status).toBe("revoked");
events.length = 0;
await db.transaction(async tx => {
await githubChatManagementService(tx as unknown as typeof db, f.providerFetch)
.linkObservedIdentity(f.endpoint.id, "owner-user", verified);
expect(events).toEqual(["intent"]);
expect((await link()).status).toBe("revoked");
});
expect(events).toEqual(["intent", "settled"]);
expect((await link()).status).toBe("linked");
events.length = 0;
await f.management.linkObservedIdentity(f.endpoint.id, "owner-user", verified);
expect(events).toEqual([]);
} finally { unsubscribe(); }
});
it("wakes milestone authority only when GitHub guest settings change, after outer commit", async () => {
const f = await reviewBotFixture();
const initial = await f.management.configuration(f.endpoint.id, "owner-user");
@@ -5550,7 +5585,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
sponsorUserId: "owner-user", permissionProfile: "restricted" as const, automaticReviews: false,
}] };
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => events.push(event));
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, (event, scopes) => {
if ([...scopes ?? []].some(scope => JSON.parse(scope).companyId === f.companyId)) events.push(event);
});
try {
await db.transaction(async tx => {
await githubChatManagementService(tx as unknown as typeof db, f.providerFetch)
@@ -52652,6 +52689,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
const stackId = randomUUID();
const f = await reopenedFixture(false, owner === "cloud_sync" ? cloudTenantPrimaryCompanyId(stackId) : undefined);
const previousToken = process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN;
const [previousUser] = owner === "cloud_sync"
? await db.select().from(authUsers).where(eq(authUsers.id, "owner-user"))
: [];
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
const notice = () => db.select().from(chatPublications).where(and(
@@ -52731,7 +52771,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
} finally {
if (previousToken === undefined) delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN;
else process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN = previousToken;
await coordinator.stop(); await f.close();
await coordinator.stop();
if (previousUser) await db.update(authUsers).set({
name: previousUser.name, email: previousUser.email,
emailVerified: previousUser.emailVerified, updatedAt: previousUser.updatedAt,
}).where(eq(authUsers.id, previousUser.id));
await f.close();
}
},
);
@@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import {
subscribeDatabaseWork,
costEvents,
activityLog,
agents,
@@ -26,6 +27,7 @@ const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() }));
vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient }));
import { heartbeatService } from "../services/heartbeat.ts";
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
import { recoveryService } from "../services/recovery/service.ts";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STALE_LOCK_TEST_DATABASE_URL");
@@ -588,8 +590,15 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => {
it.each(["observed", "workspace_finalizing", "ready_for_assessment", "arbitrating", "retryable_failure"])(
"preserves a completed native result in %s after its provider exits", async phase => {
const { runningRunId, issueId } = await seedPendingNativeResult(phase);
const result = await recoveryService(db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks();
expect(result).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [] });
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => events.push(event));
try {
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
for (let pass = 0; pass < 3; pass++) {
expect(await recovery.sweepStaleIssueLocks()).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [] });
}
expect(events).toEqual([]);
} finally { unsubscribe(); }
expect(await db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }).from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runningRunId))).toEqual([{ status: "running", errorCode: null }]);
expect(await db.select({ executionRunId: issues.executionRunId }).from(issues)
@@ -173,7 +173,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id));
const onError = vi.fn();
const maintenance = createChatReconciliationCoordinator({
processPendingDeliveries: async () => undefined, projectRunMilestones: async () => 0,
processPendingDeliveries: async () => undefined, projectNativeProgress: async () => 0,
flushPublications: async () => undefined, onError,
});
try {
@@ -57,6 +57,7 @@ function createFakeDb(options: {
chain.then = (resolve: (v: unknown) => unknown, reject?: (error: unknown) => unknown) =>
Promise.resolve().then(options.beforeWrite).then(resolve, reject);
const db = {
transaction: async (work: (tx: Db) => Promise<unknown>) => work(db),
insert: (table: unknown) => {
insertedTables.push(table);
return chain;
@@ -69,9 +70,9 @@ function createFakeDb(options: {
return {
from: (table: unknown) => ({
where: (condition: unknown) => {
// Role-default grant seeding reads existing grants first; keep that
// read working so selectThrows exercises only the degradable reads.
if (options.selectThrows && table !== principalPermissionGrants) {
// Only settings lookup is degradable. Membership locking and
// permission seeding must still work in the fail-closed flag test.
if (options.selectThrows && table === instanceSettings) {
throw new Error("select unavailable");
}
selectWheres.push({ table, condition });
@@ -81,9 +82,19 @@ function createFakeDb(options: {
: table === companyMemberships
? (options.membershipQueryRows ?? [])
: [];
return {
then: (resolve: (v: unknown) => unknown) => Promise.resolve(rows).then(resolve),
let locked = false;
const result = {
for: () => { locked = true; return result; },
then: (resolve: (v: unknown) => unknown, reject?: (error: unknown) => unknown) => {
// The optional cross-company read can fail independently of
// the primary membership transaction, which must succeed.
const query = options.selectThrows && table === companyMemberships && !locked
? Promise.reject(new Error("select unavailable"))
: Promise.resolve(rows);
return query.then(resolve, reject);
},
};
return result;
},
}),
};
@@ -329,11 +340,13 @@ describe("resolveCloudTenantActor (shared-pool hardening)", () => {
const { db, selectWheres } = createFakeDb();
await resolveCloudTenantActor(db, fakeReq(VALID_HEADERS));
const membershipSelects = selectWheres.filter((entry) => entry.table === companyMemberships);
expect(membershipSelects).toHaveLength(1);
expect(membershipSelects).toHaveLength(2);
// The first read locks the primary membership before its upsert; the
// second loads the user's active memberships for request authorization.
// Rows for other users and rows in any non-active status are excluded
// by the query itself: the filter binds exactly this user id and the
// "active" status — the same condition the session path applies.
expect(membershipSelects[0]!.condition).toEqual(
expect(membershipSelects[1]!.condition).toEqual(
and(
eq(companyMemberships.principalType, "user"),
eq(companyMemberships.principalId, "user-123"),
@@ -332,6 +332,9 @@ export function githubChatManagementService(db: Db, fetchImpl = fetch) {
expiresAt: null,
updatedAt: new Date(),
};
if (prior?.status !== "linked") {
await notifyChatRunMilestones(tx, { companyId: bot.companyId });
}
await tx
.insert(chatIdentityLinks)
.values(values)
+56 -52
View File
@@ -6158,7 +6158,61 @@ export function recoveryService(
await deps.beforeOrphanedRunTerminalWrite?.(run.id);
const now = new Date();
const updated = await db.transaction(async tx => {
const boundIssueId = boundRunIssueId(run);
const eligible = and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.status, "running"),
eq(heartbeatRuns.runtimeMode, run.runtimeMode),
nativeRunnerOwnershipNotHeldCondition(),
// Provider exit is expected while the native coordinator resumes a
// session or copies its completed workspace back. The coordinator
// owns those retries, including expired leases and future attempts.
// Check at the write so a newly recorded result cannot be orphaned
// using the earlier liveness snapshot. Terminal issue status remains
// the stronger authority.
!issueTerminalStatus && run.runtimeMode === "native"
? notExists(tx.select({ runId: nativeRunFinalizations.runId })
.from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.runId, heartbeatRuns.id),
eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId),
or(
and(
isNotNull(nativeRunFinalizations.resultId),
inArray(nativeRunFinalizations.phase, [
"observed", "workspace_finalizing", "ready_for_assessment",
"arbitrating", "retryable_failure",
]),
),
and(
isNull(nativeRunFinalizations.resultId),
or(
eq(nativeRunFinalizations.phase, "retryable_failure"),
and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)),
),
),
),
)))
: undefined,
// Recheck ownership in the write: a controller can renew or claim
// the run after the liveness read. An old snapshot cannot end a new
// controller's run, even if that controller's lease later expires.
run.runtimeMode === "legacy"
? and(
run.controllerBootId
? eq(heartbeatRuns.controllerBootId, run.controllerBootId)
: isNull(heartbeatRuns.controllerBootId),
or(
isNull(heartbeatRuns.controllerBootId),
sql`${heartbeatRuns.controllerLeaseExpiresAt} <= clock_timestamp()`,
),
)
: undefined,
);
// Do not wake projection for a retained native finalization or a run
// whose execution owner changed. Keep the same predicate on the write
// to recheck finalization records committed after this lock was taken.
const [candidate] = await tx.select().from(heartbeatRuns).where(eligible).for("update");
if (!candidate) return null;
const boundIssueId = boundRunIssueId(candidate);
if (boundIssueId) {
await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: boundIssueId });
}
@@ -6173,57 +6227,7 @@ export function recoveryService(
(terminalStatus === "interrupted" ? errorCode : null),
updatedAt: now,
})
.where(
and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.status, "running"),
eq(heartbeatRuns.runtimeMode, run.runtimeMode),
nativeRunnerOwnershipNotHeldCondition(),
// Provider exit is expected while the native coordinator resumes a
// session or copies its completed workspace back. The coordinator
// owns those retries, including expired leases and future attempts.
// Check at the write so a newly recorded result cannot be orphaned
// using the earlier liveness snapshot. Terminal issue status remains
// the stronger authority.
!issueTerminalStatus && run.runtimeMode === "native"
? notExists(tx.select({ runId: nativeRunFinalizations.runId })
.from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.runId, heartbeatRuns.id),
eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId),
or(
and(
isNotNull(nativeRunFinalizations.resultId),
inArray(nativeRunFinalizations.phase, [
"observed", "workspace_finalizing", "ready_for_assessment",
"arbitrating", "retryable_failure",
]),
),
and(
isNull(nativeRunFinalizations.resultId),
or(
eq(nativeRunFinalizations.phase, "retryable_failure"),
and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)),
),
),
),
)))
: undefined,
// Recheck ownership in the write: a controller can renew or claim
// the run after the liveness read. An old snapshot cannot end a new
// controller's run, even if that controller's lease later expires.
run.runtimeMode === "legacy"
? and(
run.controllerBootId
? eq(heartbeatRuns.controllerBootId, run.controllerBootId)
: isNull(heartbeatRuns.controllerBootId),
or(
isNull(heartbeatRuns.controllerBootId),
sql`${heartbeatRuns.controllerLeaseExpiresAt} <= clock_timestamp()`,
),
)
: undefined,
),
)
.where(eligible)
.returning()
.then((rows) => rows[0] ?? null);
});