refactor(chat): extract failed-run retry coordination

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-11 06:07:02 -05:00
1 parent a77681d39c
commit c25f0c3237
4 files changed
+2095 -1812

No files matched your search

@@ -182,3 +182,54 @@ by the service. No state or registration moves in this slice.
- Existing CI fixtures cover exact outbound ownership, interleaved control
messages, consumed progress, interactions, committed failure replacement and
issue-to-run lock order. Local database capacity remains blocked.
### Slice 5: failed chat run retry coordination
Extract source validation, native failure evidence, retry preparation, claims,
wake authorization and retry notices into `failed-run-retry.ts`. Keep the
in-flight promise map, notice cursor, error class and two authority registrations
inside the factory so each service retains its own lifetime. Keep the live
shutdown and heartbeat callbacks. The service still owns cleanup order and waits
for the module's pending retries before unregistering authorities.
- Source: 36,393 lines / 1,400,636 bytes → 34,606 lines / 1,336,443 bytes.
Net reduction: 1,787 lines / 64,193 bytes.
- Destination: 1,936 lines / 68,329 bytes.
- Baseline: all 52 checks and 5/5 review passed on slice 4 (`166ea04ac6`, #15844).
- Added three tests through the existing public service before the move. All
three passed before and after: duplicate calls share a promise, shutdown
during a pending read prevents a new claim and waits for completion, rejected
work permits a later attempt, and old-service cleanup preserves replacement
authority. These tests use controlled reads and do not require PostgreSQL.
- All 21 moved declarations and every remaining statement match the predecessor
after restoring the documented live callbacks and pending-task accessor.
Server typecheck, module boundaries and whitespace checks pass.
- Existing CI database fixtures cover corrected/edited sources, lifecycle epochs,
native recovery evidence, recreated services and concurrent retries. Local
full database execution remains blocked by the host's shared-memory limit.
Current-head full CI, broad local gates and review results belong in the PR.
## Reassessment after five slices
The service falls from 39,155 lines / 1,509,090 bytes at the rebased baseline to
34,606 lines / 1,336,443 bytes: 4,549 lines and 172,647 bytes removed (11.6% of
source lines). The five modules total 4,857 lines / 185,635 bytes, including their
imports and typed boundaries. This is useful separation, not completion of the
larger decomposition.
The service function still spans 32,254 lines. Its 244 direct nested functions
occupy 31,525 lines; 48 other statements occupy 435 lines of state, wiring,
registration and public API. The remaining lines are whitespace, comments and
syntax. The largest functions are `processMessage` (2,166), `handleAction`
(1,074), `configureWithCredentialLease` (868), `processSelectedPublication` (729),
`handleModalSubmit` (708), `handleWebhook` (685), `processLifecycleDelivery` (657)
and `applyProviderLifecycleEffect` (593).
The best next planning candidate is inbound message admission and persistence,
starting from `processMessage` and its conversation/drain ownership. Define the
transaction, live-message retention and scheduling contracts before moving it;
do not move that entire function and claim its complexity is resolved. Action
resolution and endpoint configuration are later candidates. Keep runtime lease
acquisition, partial startup and shutdown together until their ownership contract
is explicit. This batch ends with five stacked PRs; merging is separate work.
@@ -0,0 +1,83 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { chatActions, type Db } from "@paperclipai/db";
import { chatChannelService, type ChatChannelService, type ChatChannelServiceOptions } from "../services/chat-channels.js";
import { authorizeCommittedChatResponse, CommittedChatResponseAuthorizationError } from "../services/durable-chat-wakeup.js";
const services = new Set<ChatChannelService>();
afterEach(async () => {
for (const service of services) await service.shutdown();
services.clear();
});
function createService(db: Db) {
const wakeup = vi.fn().mockResolvedValue(undefined);
const runtimeShutdown = vi.fn().mockResolvedValue(undefined);
const service = chatChannelService(db, {
heartbeat: { wakeup },
runtime: { shutdown: runtimeShutdown } as unknown as ChatChannelServiceOptions["runtime"],
});
services.add(service);
return { service, wakeup, runtimeShutdown };
}
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => { resolve = done; });
return { promise, resolve };
}
describe("failed chat retry service lifetime", () => {
it("shares an in-flight retry and observes shutdown after its candidate read", async () => {
const candidate = deferred<unknown[]>();
const reads = vi.fn((table: unknown) => table === chatActions ? candidate.promise : Promise.resolve([]));
const update = vi.fn(() => { throw new Error("A shutdown retry must not claim work"); });
const db = {
select: () => ({ from: (table: unknown) => ({ where: () => reads(table) }) }),
update,
} as unknown as Db;
const { service, wakeup, runtimeShutdown } = createService(db);
const first = service.processFailedChatRunRetry("retry-action");
const second = service.processFailedChatRunRetry("retry-action");
expect(second).toBe(first);
expect(reads).toHaveBeenCalledTimes(1);
const shutdown = service.shutdown();
expect(runtimeShutdown).not.toHaveBeenCalled();
candidate.resolve([{
id: "retry-action", companyId: "company", status: "issued",
payload: { issueId: "issue" }, result: {}, updatedAt: new Date(0),
}]);
await expect(first).resolves.toMatchObject({ actionId: "retry-action", status: "queued" });
await shutdown;
services.delete(service);
expect(update).not.toHaveBeenCalled();
expect(wakeup).not.toHaveBeenCalled();
expect(runtimeShutdown).toHaveBeenCalledTimes(1);
});
it("releases a rejected in-flight retry so a later attempt can read again", async () => {
const where = vi.fn().mockResolvedValue([]);
const db = { select: () => ({ from: () => ({ where }) }) } as unknown as Db;
const { service } = createService(db);
const first = service.processFailedChatRunRetry("missing-action");
await expect(first).rejects.toMatchObject({ status: 404 });
const second = service.processFailedChatRunRetry("missing-action");
expect(second).not.toBe(first);
await expect(second).rejects.toMatchObject({ status: 404 });
expect(where).toHaveBeenCalledTimes(2);
});
it("does not unregister a replacement service's committed-response authority", async () => {
const db = {} as Db;
const first = createService(db).service;
const replacement = createService(db).service;
await first.shutdown();
services.delete(first);
const readFailure = new Error("replacement authority reached current source reads");
const tx = { select: () => { throw readFailure; } } as unknown as Db;
const input = { companyId: "company", issueId: "issue", agentId: "agent", runId: "run", resultId: "result" };
await expect(authorizeCommittedChatResponse(db, tx, input)).rejects.toBe(readFailure);
await replacement.shutdown();
services.delete(replacement);
await expect(authorizeCommittedChatResponse(db, tx, input)).rejects.toBeInstanceOf(CommittedChatResponseAuthorizationError);
});
});
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff