refactor(server): separate chat lifecycle and native progress projection

Preserve combined reconciliation and its shared budget. Retain PostgreSQL timestamp precision so rejected lifecycle candidates cannot pin pagination.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 06:37:14 -05:00
1 parent d99b2df52a
commit 362bf75b8f
3 files changed
+175 -42

No files matched your search

@@ -2,7 +2,8 @@
Baseline: Teams attempt recovery #17592 at f53ae3d6d2.
Branch: codex/sleep-28-chat-run-milestones.
Status: initial owner inspection; producer audit, implementation and acceptance pending.
Status: projection owners separated and lifecycle cursor regression repaired;
transaction producer coverage, worker registration and final acceptance pending.
## Separate obligations discovered
@@ -67,3 +68,80 @@ company fairness, bounded work and existing publication deduplication.
All earlier stack remote gates, #17490 E2E cancellation, #17580 unresolved cleanup
policy, and final master reconciliation remain required. No merge authority or
full-goal completion is implied by this audit.
## Projection boundary checkpoint
`enqueueChatRunLifecycleMilestones` now owns only coarse lifecycle projection.
`enqueueSafeNativeChatProgress` exposes the existing native progress owner.
The existing `enqueueChatRunMilestones` entry point still combines them with
lifecycle priority, the same 24-hour default window, and one capped insertion
budget. Production still calls that combined entry point. No periodic owner has
been removed prematurely; this checkpoint is not completion of slice 28.
The lifecycle cursor now carries PostgreSQL's timestamp text through a typed
`(updated_at, run_id, conversation_id)` tuple comparison. With 30 unauthorized
candidates at a submillisecond timestamp, the old Date cursor repeated a page
indefinitely, never reaching valid notices. The test caps candidate page reads
so reproducing the defect cannot leave an asynchronous SQL loop after a timeout.
Native progress has its own timestamp cursor; its precision audit belongs to
that next migration and is still required.
### Producer coverage still to implement
The current readers establish these dependencies. This table identifies signal
owners to audit; it does not claim those signals are already installed.
| Dependency | Owning paths and required wake |
| --- | --- |
| Run admission/status/issue binding/error code | `heartbeat/queue.ts`, `heartbeat/run-lifecycle.ts`, and native/legacy recovery writers. Existing `notifyExecutionRunWork` covers only some paths. Audit raw writes before adding milestone coverage. |
| Presentation decision and native result | `heartbeat/run-completion.ts`, native finalizers, and semantic receipt writers. Succeeded alone is insufficient: presentation may settle later. Source monitor receipts can authorize a successor without changing the successor row. |
| Question response claim and terminal receipt | `question-response-delivery.ts`: claim moves to delivering; `recordTerminal` owns a transaction that records target run/status. A late receipt can redirect an existing run's publication destination. Lease timestamp renewal is not a new presentation obligation. |
| Interaction state and source-run ownership | `issue-thread-interactions.ts`, plus recovery writers. Pending agent-authored questions suppress generic finals even before publication; resolved prompts can retain that suppression. |
| Published prompts, final comments, and close controls | Publication settlement and `issues.ts`. Published prompt receipts enable lineage; authored finals suppress generic completion; a published close requires fresh independently admitted sources before later presentation. Publication notifications must avoid a global scan feedback loop. |
| Inbound comment links and conversation binding | Ingress and conversation owners. The run's source string alone grants no authority. Endpoint assignment, publication mode, conversation state/issue and verified comment links matter. |
| Continuation wake/decision/effect receipts | `native-continuation-delivery.ts` reads applied status decisions, delivered continuation effects, the original wake and dispatched wake, plus responsible-user continuity. Audit their writers; a run notification before these commits is too early. |
| Monitor wake and source semantic receipt | `native-monitor-delivery.ts` requires an admitted monitor wake plus the source run's exact saved `set_task_monitor` receipt. Audit both writers, including late completion. |
| Board reply identity and provider configuration | `slackBoardReplyBindings`, endpoint configuration, and upgraded GitHub configuration affect selection. Scope configuration invalidation to its company/endpoints; preserve each provider's existing policy. |
`resolveChatOriginPublicationBindings` also follows arbitrarily long verified
lineage with cycle rejection. Changing an ancestor can affect a successor; a
strict run-ID-only invalidation is insufficient for all producer families.
Issue-scoped reconstruction is appropriate for lineage changes. Run-only hints
can be used only where the writer cannot affect descendants. Shared scheduling
must not decode provider context or implement this authorization graph.
The coarse projector currently waits on its writes and does not use skip-locked
selection. A zero insertion count may be an authoritative suppression or a
concurrent idempotent insert; database failures must retain retry. Native progress
has different lock/contention behavior and must not inherit that conclusion.
### Verification and failure record
- Initial owner split: nine selected PostgreSQL integration cases passed,
including separate-owner/combined-budget checks and six provider/surface cases.
- First cursor reproduction incorrectly guarded every `db.select`, including
legitimate authorization reads, and ran against retained earlier fixtures.
Both precision variants failed that overly broad guard; this was not evidence
of the production cursor defect. Log: `/tmp/sleep28-precision-before.log`.
- Corrected guard counts candidate-page reads. In a fresh database the millisecond
case passed; the submillisecond case failed after 100 repeating pages. Log:
`/tmp/sleep28-precision-clean-before.log`.
- After retaining PostgreSQL precision, ten selected integration cases passed in
a fresh database. Log: `/tmp/sleep28-projection-fixed-tests.log`.
- Broader final selection: 31 integration cases passed (1,398 unselected),
covering milestone authority, native progress, lock contention, question
continuations and authored finals. Log: `/tmp/sleep28-projection-acceptance.log`.
These overlap the earlier ten and are not additive.
- All 20 publication policy unit cases passed; server production typecheck passed.
Logs: `/tmp/sleep28-publication-unit.log`, `/tmp/sleep28-server-typecheck.log`.
Server typecheck excludes test files. No full-workspace acceptance or new PR
readiness is claimed for this preparatory checkpoint.
Parent #17592 at `f53ae3d6d27c2642e7dfe647a2c0a9bbf2d582ac` now has passing
Actions/security/Greptile gates, no inline review findings, and a neutral
Paperclip Review result. Neutral is not a completed positive review. #17590 at
`e501d6c99d721d6428f8c490206417765c76f4f6` has passing gates with Paperclip Review
queued at inspection. #17586 also has a neutral Paperclip Review. Final review
acceptance remains separate. Refreshed master is `2e47c42fad`; stack reconciliation
remains required. No stack PR was merged.
@@ -186,6 +186,8 @@ import {
} from "../services/chat-interaction-publications.js";
import {
enqueueChatRunMilestones,
enqueueChatRunLifecycleMilestones,
enqueueSafeNativeChatProgress,
githubRunReplyState,
resolveChatRunPresentationAuthorizationReason,
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON,
@@ -34438,7 +34440,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
]);
});
it("drains bounded Slack chat-origin milestones without admitting an internal issue run", async () => {
it.each(["milliseconds", "microseconds"])("drains bounded Slack chat-origin milestones without admitting an internal issue run (%s)", async (precision) => {
const fixture = await seedCompany();
const { callbacks, endpoint } = await configuredSlackEndpoint(fixture);
const thread = makeThread({
@@ -34533,15 +34535,32 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
})),
]);
const inserted: number[] = [];
for (let index = 0; index < 4; index += 1) {
inserted.push(
await enqueueChatRunMilestones(db, {
since: new Date("2026-09-05T14:00:00.000Z"),
limit: 1,
}),
);
if (precision === "microseconds") {
await db.update(heartbeatRuns).set({
updatedAt: sql`'2026-09-05T14:59:00.000123Z'::timestamptz`,
}).where(inArray(heartbeatRuns.id, resolverRejectedRuns.map(run => run.id)));
}
// Fail a non-advancing pagination loop explicitly instead of leaving a
// background query chain running after the test timeout.
const select = db.select.bind(db);
let reads = 0;
const readGuard = vi.spyOn(db, "select").mockImplementation((...args) => {
if (args[0] && "runUpdatedAt" in args[0] && ++reads > 100) {
throw new Error("Milestone projection did not advance past rejected candidates");
}
return select(...args);
});
const inserted: number[] = [];
try {
for (let index = 0; index < 4; index += 1) {
inserted.push(
await enqueueChatRunLifecycleMilestones(db, {
since: new Date("2026-09-05T14:00:00.000Z"),
limit: 1,
}),
);
}
} finally { readGuard.mockRestore(); }
expect(inserted).toEqual([1, 1, 1, 0]);
const publications = await db
.select()
@@ -63297,6 +63316,36 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
};
}
it.each(["separate owners", "combined budget"] as const)(
"keeps lifecycle and native progress projection independent with %s",
async (mode) => {
await enqueueChatRunMilestones(db, { since: new Date(0) });
const context = await safeNativeProgressFixture("telegram", mode === "separate owners" ? "111" : "112");
try {
const progressRun = await context.createRun("ongoing work");
await context.addEvent(progressRun);
const failedRun = await context.createRun("failed work");
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, failedRun.runId));
const input = { since: new Date(0), limit: 1 };
await expect((mode === "separate owners" ? enqueueChatRunLifecycleMilestones : enqueueChatRunMilestones)(db, input)).resolves.toBe(1);
const rows = () => db.select({ key: chatPublications.idempotencyKey }).from(chatPublications)
.where(eq(chatPublications.conversationId, context.conversation.id));
expect((await rows()).map(row => row.key)).toContain(`run:${failedRun.runId}:failed:${context.endpoint.id}`);
expect((await rows()).some(row => row.key.includes(":native:"))).toBe(false);
if (mode === "separate owners") {
await expect(enqueueChatRunLifecycleMilestones(db, input)).resolves.toBe(0);
}
await expect((mode === "separate owners" ? enqueueSafeNativeChatProgress : enqueueChatRunMilestones)(db, input)).resolves.toBe(1);
expect((await rows()).filter(row => row.key.includes(":native:"))).toEqual([
{ key: `run:${progressRun.runId}:working:${context.endpoint.id}:native:making_progress:1` },
]);
await expect(enqueueChatRunMilestones(db, input)).resolves.toBe(0);
} finally {
await retirePublicationFixture(context.service, context.endpoint.id);
}
},
);
it.each(["issue", "run"] as const)(
"skips a locked native progress %s while unrelated final and question publications deliver",
async (lockTarget) => {
+38 -32
View File
@@ -201,7 +201,7 @@ type ChatRunMilestoneCandidate = {
runId: string;
runStatus: string;
runErrorCode: string | null;
runUpdatedAt: Date;
runUpdatedAt: string;
issueId: string;
companyId: string;
endpointId: string;
@@ -280,10 +280,11 @@ export function safeMilestoneText(input: {
* lane. The selector intentionally reads only event identity, type, sequence,
* and time; native messages and payloads stay inside Paperclip.
*/
async function enqueueSafeNativeChatProgress(
export async function enqueueSafeNativeChatProgress(
db: Db,
input: { since: Date; limit: number },
options: { since?: Date; limit?: number } = {},
): Promise<number> {
const input = chatRunProjectionWindow(options);
const basePublication = alias(
chatPublications,
"safe_native_progress_base_publication",
@@ -658,16 +659,11 @@ async function enqueueSafeNativeChatProgress(
* output, errors, tool events, and reasoning stay in Paperclip. Idempotency is
* keyed by run, milestone, and endpoint so polling and restarts are harmless.
*/
export async function enqueueChatRunMilestones(
export async function enqueueChatRunLifecycleMilestones(
db: Db,
input: {
publicBaseUrl?: string | null;
since?: Date;
limit?: number;
} = {},
input: ChatRunProjectionInput = {},
): Promise<number> {
const since = input.since ?? new Date(Date.now() - 24 * 60 * 60_000);
const limit = Math.max(1, Math.min(input.limit ?? 200, 1_000));
const { since, limit } = chatRunProjectionWindow(input);
const issueIdFromContext = boundRunIssueIdSql();
const explicitlyAuthoredCommentReason = sql<boolean>`(
${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
@@ -763,7 +759,7 @@ export async function enqueueChatRunMilestones(
)`;
let inserted = 0;
let cursor: {
updatedAt: Date;
updatedAt: string;
runId: string;
conversationId: string;
} | null = null;
@@ -780,7 +776,9 @@ export async function enqueueChatRunMilestones(
runId: heartbeatRuns.id,
runStatus: heartbeatRuns.status,
runErrorCode: heartbeatRuns.errorCode,
runUpdatedAt: heartbeatRuns.updatedAt,
// Keep PostgreSQL precision: a Date cursor can repeatedly select a
// rejected page whose updated_at has submillisecond digits.
runUpdatedAt: sql<string>`${heartbeatRuns.updatedAt}::text`,
issueId: chatConversations.issueId,
companyId: chatConversations.companyId,
endpointId: chatConversations.endpointId,
@@ -939,19 +937,8 @@ export async function enqueueChatRunMilestones(
gte(heartbeatRuns.updatedAt, since),
isNull(chatPublications.id),
pageCursor
? or(
gt(heartbeatRuns.updatedAt, pageCursor.updatedAt),
and(
eq(heartbeatRuns.updatedAt, pageCursor.updatedAt),
or(
gt(heartbeatRuns.id, pageCursor.runId),
and(
eq(heartbeatRuns.id, pageCursor.runId),
gt(chatConversations.id, pageCursor.conversationId),
),
),
),
)
? sql`(${heartbeatRuns.updatedAt}, ${heartbeatRuns.id}, ${chatConversations.id}) >
(${pageCursor.updatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)`
: undefined,
),
)
@@ -1064,11 +1051,30 @@ export async function enqueueChatRunMilestones(
}
if (rows.length < pageSize) break;
}
if (inserted < limit) {
inserted += await enqueueSafeNativeChatProgress(db, {
since,
limit: limit - inserted,
});
}
return inserted;
}
type ChatRunProjectionInput = {
publicBaseUrl?: string | null;
since?: Date;
limit?: number;
};
function chatRunProjectionWindow(input: ChatRunProjectionInput) {
return {
since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000),
limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)),
};
}
/** Explicit combined reconciliation retains lifecycle priority and one shared
* publication budget. Background owners can reconcile the two duties separately. */
export async function enqueueChatRunMilestones(
db: Db,
input: ChatRunProjectionInput = {},
): Promise<number> {
const { since, limit } = chatRunProjectionWindow(input);
const inserted = await enqueueChatRunLifecycleMilestones(db, { ...input, since, limit });
if (inserted >= limit) return inserted;
return inserted + await enqueueSafeNativeChatProgress(db, { since, limit: limit - inserted });
}