fix(server): replace lifecycle milestone polling with scoped work

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 11:39:02 -05:00
1 parent d2ae1655c8
commit 1b0bec0b25
6 files changed
+146 -39

No files matched your search

@@ -1698,3 +1698,15 @@ stay silent. See the milestone audit for transaction boundaries and retained log
The previous pushed head 0f38d1079d has the recurring E2E shard-3 cancellation
failure; process-identity event append deadlock evidence is retained for diagnosis.
Production registration/poll removal and the entire remaining stack are still open.
Slice 28 production milestone migration: the app now registers the scoped
lifecycle worker and the periodic coordinator invokes only native progress.
The final GitHub guest/sponsor settings owner signals actual authority changes
inside its existing transaction; its missing intent reproduced before repair.
159 worker/startup/coordinator cases, four GitHub acceptance cases and all 16
assembled-app project-chat E2Es pass; server typecheck passes. The milestone
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.
@@ -2,8 +2,69 @@
Baseline: Teams attempt recovery #17592 at f53ae3d6d2.
Branch: codex/sleep-28-chat-run-milestones.
Status: scoped worker and initial real producer signals implemented and locally
verified; complete producer coverage, production registration and final acceptance pending.
Status: lifecycle worker registered in production; recurring lifecycle projection
removed. Selected local worker, authorization and assembled-app acceptance passes.
Final full-stack reconciliation, CI/review gates and older parent findings remain open.
## Current production disposition
The app registers `chatRunMilestoneWork` with its existing delivery coordinator.
The remaining periodic chat lane calls only `enqueueSafeNativeChatProgress` and
is named native progress in the coordinator and tests. The explicit/manual
combined reconciliation API remains available, with its original shared budget.
Publications created by the lifecycle worker notify the existing delivery owner.
The coordinator owns subscriptions, retained dirty scopes, retries and shutdown;
no independent timer or extra table is introduced. Cloud sleep proof is unchanged.
The audit's remaining alternatives were classified against actual reader fields:
| Alternate writer family | Disposition |
| --- | --- |
| Monitor semantic receipts | The real tool writes receipt plus monitor atomically before future admission; covered queue admission establishes the successor. |
| Retry scheduling in heartbeat/recovery/native replacement/capacity owners | Creates `scheduled_retry`, outside lifecycle selection. Covered promotion/claim introduces eligibility; coalescence counts and retry-strategy metadata do not supply lineage. |
| Native status dispatch and effect receipts | Existing issue-scoped signals cover original coalesced intent, dispatched wake/run binding and delivered effect. |
| Generic wake settlement | Ordinary queued→claimed/completed retains existing admission predicates; cancellation/skipping removes authority. Run status/queue producers already signal their corresponding run transition. No new general wake-count loop is needed. |
| Execution-wait, human-directed assignment and tree cancellation | Diagnostic skipped receipts have no run; human assignment has its distinct source; tree cancellation selects unclaimed null-run receipts. None creates an admitted monitor/interaction/native-status successor. |
| Task-retry, preserved-result and fresh-workspace receipts | Distinct task_retry/strategy metadata; actual successor run admission/finalization is covered. |
| Dot/Muse assignment follow-up receipts | Board conversation/issue-comment paths; they do not establish external lifecycle lineage merely by sharing an issue. |
| Slack Board reply bindings | Source issue.comment uses its own response path; it is excluded from lifecycle candidate sources. |
| GitHub configuration | The only configuration writer upserts; deletion is endpoint/company teardown rather than restoration. Upgrading disables automatic queued/working notices. Guest/sponsor changes can restore post-close authority and now signal before mutation in the existing locked transaction. |
| Initial local-board membership / agent compatibility backfill | Startup precedes worker reconstruction; agent-only memberships do not restore human authority. Live Cloud and board-claim human restoration now has signals. |
This table complements the real producer cases and chronological audit below;
it does not claim independent SQL writers or another replica are covered.
The source of truth remains existing rows and the original 24-hour run window.
Company hints are used only for broader configuration/authority operations;
issue hints cover lineage and its descendants. Settled projection does not retain
periodic discovery merely because history or configuration exists.
Latest local evidence:
- GitHub guest-setting transaction regression failed before its signal in
/tmp/sleep28-github-config-before.log; the production owner now emits intent
before mutation, settles after the outer commit, and stays silent for an
unchanged settings save. Four real GitHub settings/guest/authorization cases
pass in /tmp/sleep28-github-config-acceptance.log.
- All 159 cases in seven worker/lifecycle/identity/recovery/governed/resource,
app-coordinator and startup suites pass in
/tmp/sleep28-production-worker-acceptance.log. These include configured idle,
retained history, first producer, outer rollback/commit, retry, restart,
drain/resume and shutdown tests; not every case runs the full HTTP app.
- All 16 project-chat Playwright cases pass against a fresh assembled app at
port 32658, database paperclip_sleep28_production_e2e_20261025_test.
Log /tmp/sleep28-production-e2e.log; traces /tmp/sleep28-production-e2e-results.
This includes split-handoff cancellation and experiment restoration, but local
success does not resolve the observed remote cancellation deadlock.
- Server production typecheck passes in /tmp/sleep28-production-worker-typecheck.log.
The compiler excludes tests. Server build passes in
/tmp/sleep28-production-worker-build.log, and the four GitHub owner cases pass
again after the endpoint-deletion race guard in /tmp/sleep28-github-config-final.log.
No fleet suspension/cost result is claimed. Final remote gates remain required.
## Historical implementation checkpoints
The entries below preserve earlier incomplete states, failures and subsequent
repairs. Their statements that production still polls describe those checkpoints;
the current production disposition above supersedes them.
## Separate obligations discovered
@@ -25,7 +25,7 @@ describe("createChatReconciliationCoordinator", () => {
processFailedGitHubWebhookDeliveries,
processPendingDeliveries,
flushPublications,
projectRunMilestones: async () => 0,
projectNativeProgress: async () => 0,
processPendingSlackFileUploadReceipts: async () => undefined,
processPendingSlackSessionSyncs: async () => undefined,
onError,
@@ -142,14 +142,14 @@ describe("createChatReconciliationCoordinator", () => {
});
it.each([false, true])(
"isolates a blocked milestone projector and joins its completion (rejects: %s)",
"isolates a blocked native progress projector and joins its completion (rejects: %s)",
async (rejects) => {
let release!: () => void;
const held = new Promise<void>((resolve) => {
release = resolve;
});
const failure = new Error("milestone projection failed");
const projectRunMilestones = vi
const failure = new Error("native progress projection failed");
const projectNativeProgress = vi
.fn(async () => 0)
.mockImplementationOnce(async () => {
await held;
@@ -159,7 +159,7 @@ describe("createChatReconciliationCoordinator", () => {
const flushPublications = vi.fn(async () => undefined);
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
projectNativeProgress,
processPendingDeliveries: async () => undefined,
flushPublications,
processPendingSlackFileUploadReceipts: async () => undefined,
@@ -169,7 +169,7 @@ describe("createChatReconciliationCoordinator", () => {
try {
coordinator.reconcile();
await vi.waitFor(() => {
expect(projectRunMilestones).toHaveBeenCalledTimes(1);
expect(projectNativeProgress).toHaveBeenCalledTimes(1);
expect(flushPublications).toHaveBeenCalledTimes(1);
});
// A newly committed final/question must dispatch even while a different
@@ -179,7 +179,7 @@ describe("createChatReconciliationCoordinator", () => {
await vi.waitFor(() =>
expect(flushPublications).toHaveBeenCalledTimes(2),
);
expect(projectRunMilestones).toHaveBeenCalledTimes(1);
expect(projectNativeProgress).toHaveBeenCalledTimes(1);
let drained = false;
const draining = coordinator.drain().then(() => {
drained = true;
@@ -188,17 +188,17 @@ describe("createChatReconciliationCoordinator", () => {
expect(drained).toBe(false);
release();
await draining;
expect(projectRunMilestones).toHaveBeenCalledTimes(2);
expect(projectNativeProgress).toHaveBeenCalledTimes(2);
expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0);
if (rejects)
expect(onError).toHaveBeenCalledWith("run milestones", failure);
expect(onError).toHaveBeenCalledWith("native progress", failure);
const completedCounts = [
projectRunMilestones.mock.calls.length,
projectNativeProgress.mock.calls.length,
flushPublications.mock.calls.length,
];
await new Promise<void>((resolve) => setImmediate(resolve));
expect([
projectRunMilestones.mock.calls.length,
projectNativeProgress.mock.calls.length,
flushPublications.mock.calls.length,
]).toEqual(completedCounts);
} finally {
@@ -209,18 +209,18 @@ describe("createChatReconciliationCoordinator", () => {
},
);
it("joins a blocked milestone projector at shutdown without starting dirty follow-ups", async () => {
it("joins a blocked native progress projector at shutdown without starting dirty follow-ups", async () => {
let release!: () => void;
const held = new Promise<void>((resolve) => {
release = resolve;
});
const projectRunMilestones = vi.fn(async () => {
const projectNativeProgress = vi.fn(async () => {
await held;
return 1;
});
const flushPublications = vi.fn(async () => undefined);
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
projectNativeProgress,
flushPublications,
processPendingDeliveries: async () => undefined,
processPendingSlackFileUploadReceipts: async () => undefined,
@@ -230,7 +230,7 @@ describe("createChatReconciliationCoordinator", () => {
try {
coordinator.reconcile();
await vi.waitFor(() => {
expect(projectRunMilestones).toHaveBeenCalledOnce();
expect(projectNativeProgress).toHaveBeenCalledOnce();
expect(flushPublications).toHaveBeenCalledOnce();
});
coordinator.notifyPublications();
@@ -243,7 +243,7 @@ describe("createChatReconciliationCoordinator", () => {
expect(drained).toBe(false);
release();
await draining;
expect(projectRunMilestones).toHaveBeenCalledOnce();
expect(projectNativeProgress).toHaveBeenCalledOnce();
expect(flushPublications).toHaveBeenCalledOnce();
} finally {
release();
@@ -269,7 +269,7 @@ describe("createChatReconciliationCoordinator", () => {
);
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
projectNativeProgress: async () => 0,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -317,7 +317,7 @@ describe("createChatReconciliationCoordinator", () => {
const processPendingSlackSessionSyncs = vi.fn(async () => undefined);
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
projectNativeProgress: async () => 0,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -5542,6 +5542,31 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
.where(eq(chatEndpointLeases.endpointId, f.endpoint.id)),
).toHaveLength(0);
});
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");
const configuration = { ...initial.configuration, people: [{
kind: "guest" as const, githubUserId: "84", login: "contributor",
sponsorUserId: "owner-user", permissionProfile: "restricted" as const, automaticReviews: false,
}] };
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => events.push(event));
try {
await db.transaction(async tx => {
await githubChatManagementService(tx as unknown as typeof db, f.providerFetch)
.saveConfiguration(f.endpoint.id, { expectedRevision: initial.revision, configuration }, "owner-user");
expect(events).toEqual(["intent"]);
expect((await f.service.get(f.endpoint.id)).allowUnlinkedPeople).toBe(false);
});
expect(events).toEqual(["intent", "settled"]);
expect((await f.service.get(f.endpoint.id)).allowUnlinkedPeople).toBe(true);
events.length = 0;
const current = await f.management.configuration(f.endpoint.id, "owner-user");
await f.management.saveConfiguration(f.endpoint.id, { expectedRevision: current.revision, configuration }, "owner-user");
expect(events).toEqual([]);
} finally { unsubscribe(); }
});
it("scopes guest tasks and rechecks their sponsor before every bot tool", async () => {
const f = await reviewBotFixture();
const initial = await f.management.configuration(
+18 -16
View File
@@ -195,7 +195,8 @@ import { createToolGatewayService } from "./services/tool-gateway.js";
import { toolAccessService } from "./services/tool-access.js";
import { chatChannelService } from "./services/chat-channels.js";
import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js";
import { enqueueChatRunMilestones } from "./services/chat-run-publications.js";
import { enqueueSafeNativeChatProgress } from "./services/chat-run-publications.js";
import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js";
import {
createCoalescedAsyncTrigger,
isChatPublicationCommitSignal,
@@ -323,7 +324,7 @@ export function shouldEnablePrivateHostnameGuard(opts: {
type ChatReconciliationLane =
| "deliveries"
| "GitHub webhook recovery"
| "run milestones"
| "native progress"
| "publications"
| "Slack file receipts"
| "Slack session status";
@@ -336,7 +337,7 @@ type ChatReconciliationLane =
export function createChatReconciliationCoordinator(input: {
processPendingDeliveries: () => Promise<unknown>;
processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
projectRunMilestones: () => Promise<number>;
projectNativeProgress: () => Promise<number>;
flushPublications: () => Promise<unknown>;
processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
processPendingSlackSessionSyncs?: () => Promise<unknown>;
@@ -348,15 +349,15 @@ export function createChatReconciliationCoordinator(input: {
run: input.flushPublications,
onError: (error) => input.onError("publications", error),
});
const milestoneReconciliation = createCoalescedAsyncTrigger({
const progressReconciliation = createCoalescedAsyncTrigger({
run: async () => {
const inserted = await input.projectRunMilestones();
const inserted = await input.projectNativeProgress();
// Existing final/question publications never wait on this optional
// projection. Newly committed milestones get a bounded dispatch wake;
// projection. Newly committed progress gets a bounded dispatch wake;
// an empty/contended pass does not create a self-sustaining loop.
if (inserted > 0) publicationReconciliation.notify();
},
onError: (error) => input.onError("run milestones", error),
onError: (error) => input.onError("native progress", error),
});
const start = (
lane: ChatReconciliationLane,
@@ -382,24 +383,24 @@ export function createChatReconciliationCoordinator(input: {
input.processFailedGitHubWebhookDeliveries,
);
}
milestoneReconciliation.poll();
progressReconciliation.poll();
publicationReconciliation.poll();
if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs);
},
notifyPublications() {
milestoneReconciliation.notify();
progressReconciliation.notify();
publicationReconciliation.notify();
},
stop() {
stopped = true;
milestoneReconciliation.stop();
progressReconciliation.stop();
publicationReconciliation.stop();
},
async drain() {
await Promise.allSettled([
...inFlight.values(),
milestoneReconciliation.drain(),
progressReconciliation.drain(),
]);
// Projecting the final batch can notify dispatch after an earlier drain
// would have returned. Join dispatch only after its producer has drained.
@@ -1304,6 +1305,9 @@ export async function createApp(
chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatRunMilestoneWork(db, opts.authPublicBaseUrl).register(
deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(),
);
emailChannels.start();
const reconcileChatPublicationMaintenance = async () => {
await chatChannels.processPublicationMaintenance();
@@ -1314,10 +1318,7 @@ export async function createApp(
processPendingDeliveries: async () => undefined,
processFailedGitHubWebhookDeliveries: () =>
chatChannels.processFailedGitHubWebhookDeliveries(),
projectRunMilestones: () =>
enqueueChatRunMilestones(db, {
publicBaseUrl: opts.authPublicBaseUrl,
}),
projectNativeProgress: () => enqueueSafeNativeChatProgress(db),
flushPublications: () => reconcileChatPublicationMaintenance(),
onError: (lane, err) => {
logger.error({ err, lane }, `Failed to reconcile chat ${lane}`);
@@ -1329,7 +1330,8 @@ export async function createApp(
chatReconciliation.notifyPublications();
},
);
// Provider/action maintenance and milestone projection still use this cadence.
// Native progress and remaining publication maintenance still use this cadence.
// Lifecycle milestones use committed changes and their own recovery worker.
// Inbox, outbox, and file receipt queues have independent commit/deadline wakes.
let chatPublicationTimer: ReturnType<typeof setInterval> | null = setInterval(
() => {
+10 -3
View File
@@ -23,6 +23,7 @@ import {
import { badRequest, conflict, forbidden, notFound } from "../errors.js";
import { toolAccessService } from "./tool-access.js";
import { logActivity } from "./activity-log.js";
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
import { githubBotRequest } from "./chat-github-client.js";
import { environmentService } from "./environments.js";
import { instanceSettingsService } from "./instance-settings.js";
@@ -142,11 +143,12 @@ export function githubChatManagementService(db: Db, fetchImpl = fetch) {
"Repository overrides must refer to this App installation",
);
await db.transaction(async (tx) => {
await tx
.select({ id: chatEndpoints.id })
const [currentEndpoint] = await tx
.select({ allowUnlinkedPeople: chatEndpoints.allowUnlinkedPeople, sponsorUserId: chatEndpoints.sponsorUserId })
.from(chatEndpoints)
.where(eq(chatEndpoints.id, endpointId))
.for("update");
if (!currentEndpoint) throw notFound("GitHub bot not found");
await tx.execute(
sql`select pg_advisory_xact_lock(hashtextextended(${`github-config:${endpointId}`}, 0))`,
);
@@ -157,6 +159,11 @@ export function githubChatManagementService(db: Db, fetchImpl = fetch) {
.for("update");
if ((current?.revision ?? 0) !== parsed.expectedRevision)
throw conflict("This configuration changed. Refresh before saving.");
const allowUnlinkedPeople = config.people.some((p) => p.kind === "guest");
if ((!currentEndpoint.allowUnlinkedPeople && allowUnlinkedPeople) ||
(allowUnlinkedPeople && currentEndpoint.sponsorUserId !== config.responsibleUserId)) {
await notifyChatRunMilestones(tx, { companyId: bot.companyId });
}
const values = {
companyId: bot.companyId,
endpointId,
@@ -177,7 +184,7 @@ export function githubChatManagementService(db: Db, fetchImpl = fetch) {
await tx
.update(chatEndpoints)
.set({
allowUnlinkedPeople: config.people.some((p) => p.kind === "guest"),
allowUnlinkedPeople,
sponsorUserId: config.responsibleUserId,
updatedAt: new Date(),
})