mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
fix(server): wake native chat progress from committed work
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
b212886949
commit
bcf2f38ff5
17 files changed
+711
-281
No files matched your search
@@ -1719,3 +1719,20 @@ 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.
|
||||
|
||||
## Slice 29: native chat progress implementation
|
||||
|
||||
Branch `codex/sleep-29-native-chat-progress`, parent #17606 at `ed76256b97`.
|
||||
Native event producers and existing eligibility notifications now wake a scoped
|
||||
worker. Progress leaves the chat timer; retained history becomes quiet. The
|
||||
worker retains only unfinished batches, concrete lock contention and failures.
|
||||
See [the owner audit](2026-10-25-native-chat-progress-sleep-audit.md) for producer
|
||||
coverage, cursor precision repair, authorization boundaries and retained failures.
|
||||
|
||||
Forty-four focused native-progress cases pass, as do 13 app coordinator cases and
|
||||
the server build/typecheck. All 16 browser chat cases pass before the final
|
||||
base-publication hook. The hook fixes a reproduced commit-order wake gap; its
|
||||
19 worker and 13 coordinator cases pass. Final browser acceptance and
|
||||
publication/checks are pending.
|
||||
Parent #17606 now has passing checks and Greptile 5/5 on `ed76256b97`; all three
|
||||
review threads are resolved. This is not completion of the broader stack.
|
||||
@@ -2,141 +2,145 @@
|
||||
|
||||
Parent: #17606 at `ed76256b979541b962fa784016a550c32920d670`.
|
||||
Branch: `codex/sleep-29-native-chat-progress`.
|
||||
Status: groundwork verified; production still polls. Do not mark migrated.
|
||||
Status: implementation and focused PostgreSQL acceptance complete; remote acceptance pending; browser acceptance passed before the final base-publication hook. This is not a deployed sleep claim.
|
||||
|
||||
## Duty and owner
|
||||
|
||||
Native progress is the separate `projectNativeProgress` lane in the one-second
|
||||
chat reconciliation coordinator. It projects a closed allowlist of durable run
|
||||
event types into coarse provider-safe phrases. It never copies event messages,
|
||||
payloads, tool names, arguments, or results into external publications.
|
||||
Native progress used the one-second chat reconciliation cadence. It projects a
|
||||
closed allowlist of durable run event types into coarse provider-safe phrases.
|
||||
It never copies event messages, payloads, tool names, arguments, or results into
|
||||
external publications.
|
||||
|
||||
`chat-native-progress.ts` owns selection, authority checks and insertion.
|
||||
`chat-native-progress.ts` owns selection, authorization and insertion.
|
||||
`safe-native-chat-progress.ts` owns the normalized allowlist and phrase policy.
|
||||
`chat-run-publication-policy.ts` holds only the existing projection window and
|
||||
shared automatic-publication eligibility. The latter preserves the distinction
|
||||
between upgraded GitHub bots (tool-owned replies) and legacy automatic replies.
|
||||
The original export and explicit combined lifecycle/progress API stay compatible.
|
||||
No generic scheduler should interpret provider payloads or decide disclosure.
|
||||
`chat-run-publication-policy.ts` holds the existing projection window and shared
|
||||
automatic-publication eligibility. It preserves upgraded GitHub bots' tool-owned
|
||||
replies and legacy automatic replies. The original export and explicit combined
|
||||
lifecycle/progress API stay compatible, including their shared insertion budget.
|
||||
Startup registers the worker; it no longer invokes progress on the chat timer.
|
||||
Other chat maintenance duties remain for their own migration slices.
|
||||
|
||||
## Existing durable contract
|
||||
## Durable contract and triggers
|
||||
|
||||
- Native running run, matching company/agent and bound issue.
|
||||
- Active/waiting conversation with an automatic endpoint assigned to that agent.
|
||||
- Existing base working publication in pending/retry/streaming/published state.
|
||||
- Latest allowlisted event for that run, inside the existing 24-hour window.
|
||||
- Event time at least 20 seconds after the base working publication.
|
||||
- Exact current origin/lineage authorization; current provider interaction and
|
||||
explicitly authored final output may suppress this optional progress.
|
||||
- At most one publication per phase. Twenty-second spacing compares the saved
|
||||
event time with saved publication times. Passage of wall-clock time does not
|
||||
make the same suppressed old event eligible. Do not invent a timer to retry it
|
||||
after twenty seconds or change the cadence policy as part of this migration.
|
||||
- Existing publications provide durable deduplication and outbound delivery.
|
||||
Do not introduce another queue table.
|
||||
The candidate requires a native running run, matching company/agent and bound
|
||||
issue, an active/waiting conversation, and an automatic endpoint assigned to the
|
||||
agent. Its existing base working publication must be pending, retry, streaming,
|
||||
or published. The latest allowlisted event must be within the existing 24-hour
|
||||
window and at least 20 seconds after the base publication. Existing lineage,
|
||||
interaction arbitration, final-output suppression and delivery authorization stay
|
||||
with their current owners. The ordinary direct-origin path and stricter
|
||||
post-close lineage path retain their existing different authorization stages.
|
||||
|
||||
## Verified pagination repair
|
||||
There is at most one publication per phase. Spacing compares saved event time
|
||||
with saved publication times. Waiting twenty seconds does not make an old event
|
||||
eligible. No timer retries cadence-suppressed events. Existing events and
|
||||
publications are durable truth; there is no new queue table or schema migration.
|
||||
|
||||
The selector previously round-tripped event timestamps through JavaScript Date.
|
||||
PostgreSQL microseconds were lost. With a full page of rejected candidates at
|
||||
one fractional timestamp, every next page selected those same candidates; an
|
||||
eligible event later in the ordered stream was never reached.
|
||||
The two event writers register awaited intent before insertion, after duplicate
|
||||
and ownership checks, inside their existing transaction boundaries:
|
||||
|
||||
The new regression creates 30 origin-rejected candidates with a microsecond
|
||||
timestamp and a valid event after them. A query-count guard fails the broken
|
||||
pagination without leaving an infinite SQL chain after the test timeout.
|
||||
The fix keeps the timestamp as PostgreSQL text and uses the full timestamp/run/
|
||||
conversation tuple for continuation, preserving ordering and eligibility.
|
||||
- `appendHeartbeatRunEvent` covers native ingress through
|
||||
`PaperclipControlPlanePort`, direct allowlisted events and recovered trace diffs.
|
||||
It reads only runtime mode and bound issue alongside the existing binding
|
||||
fields. Ordinary non-progress direct log events retain their atomic-statement
|
||||
path. Direct allowlisted event writes now use a transaction even for legacy
|
||||
runs; only native bound runs signal progress.
|
||||
- `NativeRunCoordinatorStore.appendEvent` has its own insertion path. It signals
|
||||
after binding, sequence and duplicate validation, before persistence.
|
||||
- `persistReprojectedWorkspaceDiffs` delegates to the first writer and is tested
|
||||
through its actual recovered-trace persistence API, including duplicate replay.
|
||||
|
||||
- `/tmp/sleep29-progress-cursor-before.log`: fails with non-advancing pagination.
|
||||
- `/tmp/sleep29-progress-cursor-acceptance.log`: 17 selected cases pass after repair.
|
||||
- `/tmp/sleep29-progress-owner-acceptance.log`: 21 native-progress and combined
|
||||
projection cases pass after extracting the owner.
|
||||
- `/tmp/sleep29-progress-owner-typecheck.log`: server production typecheck passes.
|
||||
This excludes test files. No worker-quiet or production migration claim yet.
|
||||
- Database: isolated external PostgreSQL
|
||||
`paperclip_sleep28_review_chat_20261025_test`, through
|
||||
`PAPERCLIP_TEST_DATABASE_URL`. Tests retire their publication endpoints.
|
||||
Eligibility mutations share slice 28's issue/company notification boundary.
|
||||
`notifyChatRunMilestones` now invalidates both lifecycle and native progress,
|
||||
without adding recovery wrappers to every caller. Both topics share the outer
|
||||
transaction's identity and commit/rollback outcome. Existing writers cover
|
||||
lineage receipts, run state, interactions, endpoint/resource configuration,
|
||||
identity and membership. Failed-publication replay and operator retry now supply
|
||||
the affected issue to `notifyChatPublicationWork`: they can restore an existing
|
||||
base working publication without any new run event. The lifecycle projector also signals native progress before creating a first
|
||||
working publication. A long transaction can give that publication an older
|
||||
transaction-start timestamp but commit after an event was already inspected.
|
||||
Routine transport claim bookkeeping retains its narrower notification behavior.
|
||||
|
||||
## Producer audit and remaining implementation
|
||||
## Scheduling, contention and lifetime
|
||||
|
||||
Two event persistence owners need coverage, not just the post-commit live event:
|
||||
`chat-native-progress-work.ts` uses `registerScopedDeadlineWork` and the shared
|
||||
delivery coordinator. It reconstructs relevant issue scopes on startup, refreshes
|
||||
only the affected issue/company after a committed change, and projects batches
|
||||
of 100. A full insertion batch retains immediate work. Lock contention retains a
|
||||
one-second retry; failure uses the shared backoff. A settled short pass disarms.
|
||||
The shared worker preserves changes during execution, paused work during drain,
|
||||
startup failures through resume, fair company dispatch and shutdown ownership.
|
||||
|
||||
1. `heartbeat-run-events.ts::appendHeartbeatRunEvent`: native ingestion uses its
|
||||
transaction/deduplication path. `paperclip-control-plane-port.ts` wraps this
|
||||
in an outer transaction with active-authority checks before and after append.
|
||||
Intent must precede the event write and follow outer commit/rollback. Duplicate
|
||||
replay must not create fresh projection work. Ordinary non-progress log events
|
||||
must not acquire a transaction/XID merely for this feature.
|
||||
2. `native-run-coordinator-store.ts::appendEvent`: directly inserts run events
|
||||
after current-run and sequence/duplicate checks. It does not call the generic
|
||||
append helper. Cover it explicitly in its existing transaction.
|
||||
Projection retains issue-before-run SKIP LOCKED ordering. A skipped lock is a
|
||||
concrete obligation only while the matching row still exists and the run is
|
||||
native/running. Deleted or terminal candidates settle. Authority, latest event,
|
||||
phase deduplication and final output are checked again inside projection.
|
||||
|
||||
Additional paths found:
|
||||
The selector also repairs an existing pagination bug: JavaScript Date truncated
|
||||
PostgreSQL microseconds, repeatedly selecting a full rejected page. Cursors now
|
||||
retain PostgreSQL timestamp text and the full timestamp/run/conversation tuple.
|
||||
The regression seeds 30 origin-rejected candidates at one fractional timestamp
|
||||
and an eligible later event. A query-count guard bounds failure of the old code.
|
||||
|
||||
- `provider-trace-workspace-diff-reprojection.ts` appends
|
||||
`workspace.diff.recorded` through the generic native-source path. Recovery can
|
||||
therefore create a new eligible event independently of live runner ingress.
|
||||
- `heartbeat/run-lifecycle.ts` uses the generic writer for dynamic event types.
|
||||
Preserve its redaction/runtime-status responsibilities; do not project raw data.
|
||||
- Direct insertion in `heartbeat/queue.ts` writes a lifecycle event; it is outside
|
||||
the closed progress allowlist. Coordinator-store direct insertion is not.
|
||||
- Other native helper callers currently append lifecycle, process, cleanup, or
|
||||
credential events. Their existence is not permission to signal every log entry.
|
||||
## Evidence and failures retained
|
||||
|
||||
Eligibility changes can make a saved event actionable without a new event:
|
||||
run/issue binding and state, conversation state, current identity/membership,
|
||||
endpoint/connection/resource authority, lineage receipts, interaction arbitration,
|
||||
and base-publication state. Reuse the existing issue/company-scoped notification
|
||||
boundary audited in slice 28. Extend that shared boundary once; avoid repeated
|
||||
recovery wrappers across routes. Inspect all base-publication retry and operator
|
||||
resolution owners before removing the poll. `notifyChatPublicationWork` already
|
||||
accepts an optional issue scope for projection-affecting changes; transport claim
|
||||
bookkeeping intentionally does not invalidate projection.
|
||||
- `/tmp/sleep29-progress-cursor-before.log`: causal pagination failure.
|
||||
- `/tmp/sleep29-progress-cursor-acceptance.log`: 17 selected cases after repair.
|
||||
- `/tmp/sleep29-progress-owner-acceptance.log`: 21 projection cases after extraction.
|
||||
- `/tmp/sleep29-native-producers-before.log`: real event-writer cases fail before
|
||||
notification hooks; startup and lock cases already pass.
|
||||
- `/tmp/sleep29-native-producers-fixed.log`: all five original producer cases pass.
|
||||
- `/tmp/sleep29-native-lifecycle.log`: 14 worker lifecycle cases pass.
|
||||
- `/tmp/sleep29-progress-recovery.log`: trace recovery and 105-run backlog pass.
|
||||
Its attempted publication-mode test was invalid: endpoint update does not
|
||||
accept that field. No production API was expanded to satisfy the test.
|
||||
- `/tmp/sleep29-progress-eligibility.log`: the first identity test incorrectly
|
||||
assumed ordinary direct-origin projection enforced post-close authority.
|
||||
The replacement uses the real closed/reopened conversation and admission
|
||||
receipts, where the existing projection contract requires current authority.
|
||||
- `/tmp/sleep29-progress-eligibility-fixed.log`: five real-service cases pass:
|
||||
post-close identity, resource and visibility restoration, plus failed publication
|
||||
replay and operator retry. No new event or manual projection call is used.
|
||||
- `/tmp/sleep29-progress-worker-acceptance.log`: 13 app coordinator tests pass;
|
||||
the concurrent-event test's quiet assertion initially counted old calls from
|
||||
an already-installed transaction spy. Restoring that barrier spy before taking
|
||||
the quiet baseline corrects the test accounting, without a production change.
|
||||
- `/tmp/sleep29-final-native-acceptance.log`: 44 selected tests pass across the
|
||||
18 worker cases and 26 existing/new integration cases. These cover both real
|
||||
event owners, nested outer commit/rollback, duplicate replay, irrelevant logs,
|
||||
issue/run lock contention and terminalization, startup reconstruction, drain,
|
||||
failed startup through resume, shutdown, trace recovery, issue/company scope,
|
||||
concurrent commits, full batches, cadence, final suppression and provider-safe
|
||||
content. The name filter skips the app file; its 13 cases passed separately.
|
||||
- `/tmp/sleep29-wired-typecheck.log`: production server typecheck passes.
|
||||
- `/tmp/sleep29-server-build.log`: server build passes (production TypeScript;
|
||||
test files are executed by Vitest, not included in the server typecheck).
|
||||
- `/tmp/sleep29-first-base-before.log`: causal missing-wake failure when a
|
||||
first working publication commits after event selection. The test backdates
|
||||
the base inside its transaction to model transaction-start timestamp ordering.
|
||||
- `/tmp/sleep29-base-fixed-workers.log`: all 19 worker and 13 app coordinator
|
||||
cases pass after the lifecycle producer signals native progress.
|
||||
- Browser acceptance: all 16 `agent-chat-projects.spec.ts` cases pass, log
|
||||
`/tmp/sleep29-app-e2e.log`, fresh isolated database, port 32659. This run
|
||||
precedes the final base-publication hook; a final run remains pending.
|
||||
|
||||
## Contention and scheduling
|
||||
PostgreSQL is external and isolated because embedded PostgreSQL is unavailable
|
||||
on this host. Worker fixtures clean their own companies. The reused chat fixture
|
||||
DB retains history; tests stop workers and retire endpoints after assertions.
|
||||
Configured-idle and retained-history quiet assertions observe select/transaction
|
||||
calls, including two minutes of controlled time in worker tests. Provider APIs
|
||||
are controlled fixtures, not live Slack/Telegram/Discord accounts.
|
||||
|
||||
The current projector locks issue then run with SKIP LOCKED. This order avoids
|
||||
inversion with native finalization and publication foreign keys. A zero insert
|
||||
count currently conflates ordinary suppression, deletion/terminalization, and
|
||||
lock contention. The polling caller retries all of them later.
|
||||
## Limits and remaining acceptance
|
||||
|
||||
Before replacing that caller, expose whether concrete candidates were deferred
|
||||
by contention. Retain a retry only for unresolved candidates or failure. Recheck
|
||||
existence/current eligibility so a deleted or terminal run does not become an
|
||||
endless retry. A later committed event or eligibility change must remain dirty
|
||||
if it arrives during the current sweep.
|
||||
Process-local notifications retain the single active writer model. Independent
|
||||
SQL writers and other processes need a wake integration. No Cloud sleep gate or
|
||||
external alarm coverage changes here. An internal deadline cannot wake stopped
|
||||
compute. Browser acceptance, exact-head CI, and review remain required before
|
||||
this slice is ready. The whole stack remains unmerged.
|
||||
|
||||
Use the shared scoped-deadline worker and delivery coordinator for coalescing,
|
||||
company fairness, failure backoff, drain/resume and shutdown. Reconstruct relevant
|
||||
issue scopes at startup. Scope refresh and projection must use the same candidate
|
||||
contract. A short settled pass disarms; unfinished batches retain work. Preserve
|
||||
24-hour policy rather than broadening history during this migration.
|
||||
|
||||
## Required acceptance before removing periodic dispatch
|
||||
|
||||
- Configured idle and retained completed/suppressed history generate no recurring
|
||||
worker SQL after startup and settlement.
|
||||
- Real event producers wake after outer commit, not savepoint release; rollback
|
||||
publishes nothing and settles uncertainty through the existing protocol.
|
||||
- Duplicate native-event replay and irrelevant log entries create no new debt.
|
||||
- Both event persistence owners and trace reprojection have causal coverage.
|
||||
- Restored authority/base-publication state wakes saved eligible events.
|
||||
- Contended issue and run locks retain recovery while unrelated work proceeds;
|
||||
terminalization, final answers and changed authority win before retry.
|
||||
- Full batches, submillisecond cursor progression, cadence and phase deduplication.
|
||||
- Restart, failure, changes during a pass, drain/resume, and shutdown.
|
||||
- Assembled startup registers the worker and removes this one periodic lane;
|
||||
other chat maintenance lanes remain until their own migrations.
|
||||
- Current-head checks/review, final base reconciliation, and an accurate PR body.
|
||||
|
||||
Process-local signals retain the single active writer model. No Cloud sleep gate
|
||||
or external alarm coverage changes here; worker quiet is not provider suspension.
|
||||
|
||||
## Remote reconciliation note
|
||||
|
||||
Fetched master is `e302f2098f` (not merged into this stack yet). Newly landed
|
||||
Fetched master is `19ec787e5b` (not replayed into this stack yet). Newly landed
|
||||
#17616 expands native-harness backup expiration to continuity/quarantine evidence
|
||||
and adds its evidence signal. Its index.ts interval still exists. Audit this newer
|
||||
contract when migrating that separate retention owner; do not recreate or drop
|
||||
its preservation guarantees based on the old inventory.
|
||||
and adds its evidence signal. Its interval remains. Preserve that newer contract
|
||||
when migrating its separate retention owner.
|
||||
@@ -25,7 +25,6 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
processFailedGitHubWebhookDeliveries,
|
||||
processPendingDeliveries,
|
||||
flushPublications,
|
||||
projectNativeProgress: async () => 0,
|
||||
processPendingSlackFileUploadReceipts: async () => undefined,
|
||||
processPendingSlackSessionSyncs: async () => undefined,
|
||||
onError,
|
||||
@@ -78,6 +77,8 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals"));
|
||||
expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()");
|
||||
expect(maintenance).not.toContain("chatChannels.voice.");
|
||||
expect(maintenance).not.toContain("enqueueSafeNativeChatProgress");
|
||||
expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,");
|
||||
expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
|
||||
expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,");
|
||||
const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8");
|
||||
@@ -141,117 +142,6 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress(");
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"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("native progress projection failed");
|
||||
const projectNativeProgress = vi
|
||||
.fn(async () => 0)
|
||||
.mockImplementationOnce(async () => {
|
||||
await held;
|
||||
if (rejects) throw failure;
|
||||
return 1;
|
||||
});
|
||||
const flushPublications = vi.fn(async () => undefined);
|
||||
const onError = vi.fn();
|
||||
const coordinator = createChatReconciliationCoordinator({
|
||||
projectNativeProgress,
|
||||
processPendingDeliveries: async () => undefined,
|
||||
flushPublications,
|
||||
processPendingSlackFileUploadReceipts: async () => undefined,
|
||||
processPendingSlackSessionSyncs: async () => undefined,
|
||||
onError,
|
||||
});
|
||||
try {
|
||||
coordinator.reconcile();
|
||||
await vi.waitFor(() => {
|
||||
expect(projectNativeProgress).toHaveBeenCalledTimes(1);
|
||||
expect(flushPublications).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
// A newly committed final/question must dispatch even while a different
|
||||
// issue's projection is blocked. Repeated signals leave one dirty retry.
|
||||
coordinator.notifyPublications();
|
||||
coordinator.notifyPublications();
|
||||
await vi.waitFor(() =>
|
||||
expect(flushPublications).toHaveBeenCalledTimes(2),
|
||||
);
|
||||
expect(projectNativeProgress).toHaveBeenCalledTimes(1);
|
||||
let drained = false;
|
||||
const draining = coordinator.drain().then(() => {
|
||||
drained = true;
|
||||
});
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(drained).toBe(false);
|
||||
release();
|
||||
await draining;
|
||||
expect(projectNativeProgress).toHaveBeenCalledTimes(2);
|
||||
expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0);
|
||||
if (rejects)
|
||||
expect(onError).toHaveBeenCalledWith("native progress", failure);
|
||||
const completedCounts = [
|
||||
projectNativeProgress.mock.calls.length,
|
||||
flushPublications.mock.calls.length,
|
||||
];
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect([
|
||||
projectNativeProgress.mock.calls.length,
|
||||
flushPublications.mock.calls.length,
|
||||
]).toEqual(completedCounts);
|
||||
} finally {
|
||||
release();
|
||||
coordinator.stop();
|
||||
await coordinator.drain();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
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 projectNativeProgress = vi.fn(async () => {
|
||||
await held;
|
||||
return 1;
|
||||
});
|
||||
const flushPublications = vi.fn(async () => undefined);
|
||||
const coordinator = createChatReconciliationCoordinator({
|
||||
projectNativeProgress,
|
||||
flushPublications,
|
||||
processPendingDeliveries: async () => undefined,
|
||||
processPendingSlackFileUploadReceipts: async () => undefined,
|
||||
processPendingSlackSessionSyncs: async () => undefined,
|
||||
onError: vi.fn(),
|
||||
});
|
||||
try {
|
||||
coordinator.reconcile();
|
||||
await vi.waitFor(() => {
|
||||
expect(projectNativeProgress).toHaveBeenCalledOnce();
|
||||
expect(flushPublications).toHaveBeenCalledOnce();
|
||||
});
|
||||
coordinator.notifyPublications();
|
||||
coordinator.stop();
|
||||
let drained = false;
|
||||
const draining = coordinator.drain().then(() => {
|
||||
drained = true;
|
||||
});
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(drained).toBe(false);
|
||||
release();
|
||||
await draining;
|
||||
expect(projectNativeProgress).toHaveBeenCalledOnce();
|
||||
expect(flushPublications).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
release();
|
||||
coordinator.stop();
|
||||
await coordinator.drain();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps slow optional recovery from suppressing later publication sweeps", async () => {
|
||||
let releaseDelivery!: () => void;
|
||||
let releaseSlackStatus!: () => void;
|
||||
@@ -269,7 +159,6 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
);
|
||||
const onError = vi.fn();
|
||||
const coordinator = createChatReconciliationCoordinator({
|
||||
projectNativeProgress: async () => 0,
|
||||
processPendingDeliveries,
|
||||
flushPublications,
|
||||
processPendingSlackFileUploadReceipts,
|
||||
@@ -317,7 +206,6 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
const processPendingSlackSessionSyncs = vi.fn(async () => undefined);
|
||||
const onError = vi.fn();
|
||||
const coordinator = createChatReconciliationCoordinator({
|
||||
projectNativeProgress: async () => 0,
|
||||
processPendingDeliveries,
|
||||
flushPublications,
|
||||
processPendingSlackFileUploadReceipts,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { nativeChatProgressWork } from "../services/chat-native-progress-work.js";
|
||||
import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "../middleware/auth.js";
|
||||
import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js";
|
||||
import { accessService } from "../services/access.js";
|
||||
@@ -52781,6 +52782,57 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["identity", "resource", "visibility"] as const)(
|
||||
"wakes saved post-close native progress when %s authority returns", async owner => {
|
||||
const f = await reopenedFixture();
|
||||
const onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
const progress = () => db.select().from(chatPublications).where(and(
|
||||
eq(chatPublications.issueId, f.conversation.issueId), like(chatPublications.idempotencyKey, "%:native:%"),
|
||||
));
|
||||
try {
|
||||
await db.insert(chatPublications).values({
|
||||
companyId: f.fixture.companyId, endpointId: f.endpoint.id,
|
||||
conversationId: f.conversation.id, issueId: f.conversation.issueId,
|
||||
idempotencyKey: `run:${f.fresh.runId}:working:${f.endpoint.id}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" }, state: "published",
|
||||
createdAt: new Date(Date.now() - 60_000),
|
||||
});
|
||||
await db.insert(heartbeatRunEvents).values({
|
||||
companyId: f.fixture.companyId, agentId: f.fixture.assignedAgentId,
|
||||
runId: f.fresh.runId, seq: 1, eventType: "item.completed",
|
||||
});
|
||||
let restore: () => Promise<unknown>;
|
||||
if (owner === "identity") {
|
||||
await f.service.update(f.endpoint.id, { allowUnlinkedPeople: false }, "owner-user");
|
||||
await f.service.revokeLink(f.endpoint.id, f.fresh.action.principalId!);
|
||||
const intent = await f.service.createLinkIntent(f.endpoint.id, f.fresh.action.principalId!, 1800);
|
||||
restore = () => f.service.confirmIdentityLink(new URL(intent.confirmationUrl).searchParams.get("token")!, "owner-user");
|
||||
} else if (owner === "resource") {
|
||||
const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.id, f.conversation.id));
|
||||
expect(conversation.resourceId).not.toBeNull();
|
||||
await f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: false }], "owner-user");
|
||||
restore = () => f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: true }], "owner-user");
|
||||
} else {
|
||||
await issueService(db).update(f.conversation.issueId, { hiddenAt: new Date() });
|
||||
restore = () => issueService(db).update(f.conversation.issueId, { hiddenAt: null });
|
||||
}
|
||||
await nativeChatProgressWork(db).register(coordinator, () => true).ready;
|
||||
await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
expect(await progress()).toEqual([]);
|
||||
await restore();
|
||||
await expect.poll(progress, { timeout: 5000 }).toEqual([
|
||||
expect.objectContaining({ idempotencyKey: `run:${f.fresh.runId}:working:${f.endpoint.id}:native:making_progress:1` }),
|
||||
]);
|
||||
await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
const select = vi.spyOn(db, "select");
|
||||
try { await new Promise(resolve => setTimeout(resolve, 1200)); expect(select).not.toHaveBeenCalled(); }
|
||||
finally { select.mockRestore(); }
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { await coordinator.stop(); await f.close(); }
|
||||
},
|
||||
);
|
||||
|
||||
it("publishes working and the exact fresh final after pinned close without resurrecting the old source", async () => {
|
||||
const f = await reopenedFixture();
|
||||
try {
|
||||
@@ -63567,6 +63619,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
};
|
||||
}
|
||||
|
||||
it.each(["replay", "operator retry"] as const)(
|
||||
"wakes saved native progress when %s restores eligibility", async owner => {
|
||||
const context = await safeNativeProgressFixture("telegram", `12${["replay", "operator retry"].indexOf(owner)}`, "channel", "42", true);
|
||||
const onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
const progress = () => db.select().from(chatPublications).where(and(
|
||||
eq(chatPublications.issueId, context.conversation.issueId),
|
||||
like(chatPublications.idempotencyKey, "%:native:%"),
|
||||
));
|
||||
const settled = () => vi.waitFor(() => {
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 5000 });
|
||||
try {
|
||||
const run = await context.createRun("saved progress awaiting eligibility");
|
||||
await context.addEvent(run);
|
||||
const [base] = await db.update(chatPublications).set({
|
||||
state: owner === "replay" ? "failed" : "delivery_unknown",
|
||||
}).where(eq(chatPublications.idempotencyKey, `run:${run.runId}:working:${context.endpoint.id}`)).returning();
|
||||
const restore = () => owner === "replay"
|
||||
? context.service.replayPublication(context.endpoint.id, base.id)
|
||||
: context.service.resolvePublication(context.endpoint.id, base.id, "retry_anyway", "owner-user");
|
||||
await nativeChatProgressWork(db).register(coordinator, () => true).ready;
|
||||
await settled(); expect(await progress()).toEqual([]);
|
||||
await restore();
|
||||
await expect.poll(progress, { timeout: 5000 }).toEqual([
|
||||
expect.objectContaining({ idempotencyKey: `run:${run.runId}:working:${context.endpoint.id}:native:making_progress:1` }),
|
||||
]);
|
||||
await settled();
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await new Promise(resolve => setTimeout(resolve, 1200));
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { await coordinator.stop(); await retirePublicationFixture(context.service, context.endpoint.id); }
|
||||
},
|
||||
);
|
||||
|
||||
it("advances native progress past suppressed candidates with microsecond timestamps", async () => {
|
||||
const context = await safeNativeProgressFixture("telegram", "119");
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
import { enqueueChatRunLifecycleMilestones } from "../services/chat-run-publications.js";
|
||||
import { persistReprojectedWorkspaceDiffs, projectCodexWorkspaceDiffsFromTrace } from "../services/provider-trace-workspace-diff-reprojection.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, like, and } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { chatPublications, createDb, heartbeatRunEvents, heartbeatRuns, issues, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.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 { nativeChatProgressWork } from "../services/chat-native-progress-work.js";
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js";
|
||||
import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js";
|
||||
import type { PrpEvent } from "../vendor/paperclip-runner/index.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_PROGRESS_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("native chat progress work", () => {
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let fixtures: ReturnType<typeof createChatMilestoneFixtures>;
|
||||
let f: Awaited<ReturnType<typeof fixtures.create>>;
|
||||
let runId: string;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let canRun = true;
|
||||
const errors: unknown[] = [];
|
||||
beforeAll(async () => {
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-progress-");
|
||||
db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
fixtures = createChatMilestoneFixtures(db);
|
||||
}, 60_000);
|
||||
beforeEach(async () => {
|
||||
errors.length = 0; canRun = true; f = await fixtures.create(); runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId,
|
||||
nativeIssueId: f.issueId, runtimeMode: "native", status: "running", responsibleUserId: f.userId,
|
||||
contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId },
|
||||
});
|
||||
await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: f.endpointId,
|
||||
conversationId: f.conversationId, issueId: f.issueId, idempotencyKey: `run:${runId}:working:${f.endpointId}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" }, state: "published",
|
||||
createdAt: new Date(Date.now() - 60_000),
|
||||
});
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await coordinator?.stop(); coordinator = undefined;
|
||||
vi.useRealTimers(); vi.restoreAllMocks(); await fixtures.cleanup();
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
const publications = () => db.select().from(chatPublications).where(and(
|
||||
eq(chatPublications.issueId, f.issueId), like(chatPublications.idempotencyKey, "%:native:%"),
|
||||
));
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
await nativeChatProgressWork(db).register(coordinator, () => canRun).ready;
|
||||
}
|
||||
async function settle() {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
}
|
||||
async function quiet() {
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try { await vi.advanceTimersByTimeAsync(120_000); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); }
|
||||
finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
}
|
||||
function event(nativeSource = false, eventType = "item.completed") {
|
||||
return { companyId: f.companyId, agentId: f.agentId, runId, eventType,
|
||||
message: "PRIVATE event text", payload: { token: "PRIVATE token" },
|
||||
...(nativeSource ? { nativeSource: { sourceInstanceId: runId, sourceEventId: "event-1", sourceSeq: 1,
|
||||
protocolSchemaVersion: 1, canonicalPayload: { eventType } } } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
it.each([false, true])("wakes from a real committed event after idle (native source: %s)", async nativeSource => {
|
||||
await start(); await settle(); await quiet(); expect(await publications()).toEqual([]);
|
||||
const input = event(nativeSource);
|
||||
const save = (rollback: boolean) => db.transaction(async tx => {
|
||||
await appendHeartbeatRunEvent(tx as unknown as typeof db, input);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await publications()).toEqual([]);
|
||||
if (rollback) throw new Error("outer rollback");
|
||||
});
|
||||
await expect(save(true)).rejects.toThrow("outer rollback");
|
||||
await settle(); expect(await publications()).toEqual([]); await quiet();
|
||||
await save(false);
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await publications()).toEqual([expect.objectContaining({ payload: expect.objectContaining({ text: "Maya is making progress…" }) })]);
|
||||
}, { timeout: 5000 });
|
||||
await settle(); await quiet(); expect(errors).toEqual([]);
|
||||
if (nativeSource) {
|
||||
const signals: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind));
|
||||
try { expect((await appendHeartbeatRunEvent(db, input)).disposition).toBe("duplicate"); expect(signals).toEqual([]); }
|
||||
finally { unsubscribe(); }
|
||||
await quiet();
|
||||
}
|
||||
});
|
||||
|
||||
it("wakes a saved event when its first working publication commits after event selection", async () => {
|
||||
await db.delete(chatPublications).where(eq(chatPublications.issueId, f.issueId));
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
await start(); await settle(); expect(await publications()).toEqual([]); await quiet();
|
||||
await db.transaction(async tx => {
|
||||
expect(await enqueueChatRunLifecycleMilestones(tx as unknown as typeof db, {
|
||||
scope: { companyId: f.companyId, issueId: f.issueId },
|
||||
})).toBe(1);
|
||||
// Model a base transaction that began before the event but committed
|
||||
// afterwards, without a wall-clock delay in the suite. PostgreSQL's
|
||||
// default now() is transaction-start time, not commit time.
|
||||
await tx.update(chatPublications).set({ createdAt: new Date(Date.now() - 300_000) })
|
||||
.where(eq(chatPublications.idempotencyKey, `run:${runId}:working:${f.endpointId}`));
|
||||
await vi.advanceTimersByTimeAsync(1000); expect(await publications()).toEqual([]);
|
||||
});
|
||||
await settle(); expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("reconstructs saved progress after restart and becomes quiet with published history", async () => {
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
await start(); await settle(); expect(await publications()).toHaveLength(1); await quiet();
|
||||
await coordinator!.stop(); await start(); await settle();
|
||||
expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it.each(["issue", "run"] as const)("retries a contended %s without another event", async owner => {
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
let release!: () => void, entered!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(resolve => { entered = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
if (owner === "issue") await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
else await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
|
||||
entered(); await hold;
|
||||
});
|
||||
try {
|
||||
await ready; await start(); expect(await publications()).toEqual([]);
|
||||
expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
release(); await lock;
|
||||
await settle(); expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]);
|
||||
} finally { release(); await lock; }
|
||||
});
|
||||
|
||||
it.each(["port", "coordinator"] as const)("wakes from the real %s event owner after outer commit, with quiet replay", async kind => {
|
||||
const sessionId = randomUUID(), runnerId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, runId));
|
||||
const binding = { companyId: f.companyId, issueId: f.issueId, runId, agentId: f.agentId,
|
||||
sessionId, completionContractId: randomUUID(), completionContractSha256: "fixture",
|
||||
sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "control",
|
||||
normalizedSessionId: sessionId, runnerSourceInstanceId: runnerId,
|
||||
completionContractRevision: "1", completionContractCriterionIds: [],
|
||||
};
|
||||
const source: PrpEvent = { schema: "paperclip.prp.event.v1", schemaVersion: 1,
|
||||
sourceEventId: "event-1", sourceSeq: 1, sourceInstanceId: runnerId, sourceKind: "runner",
|
||||
runId, normalizedSessionId: sessionId, turnId: "turn-1", itemId: "item-1",
|
||||
eventType: "item.completed", priority: 1, emittedAt: new Date().toISOString(), payload: {},
|
||||
};
|
||||
const append = (client: typeof db) => (kind === "port"
|
||||
? new PaperclipControlPlanePort(client, binding)
|
||||
: new NativeRunCoordinatorStore(client, binding)).appendEvent(source);
|
||||
await start(); await settle(); await quiet();
|
||||
await expect(db.transaction(async tx => {
|
||||
await append(tx as unknown as typeof db); throw new Error("rollback owner");
|
||||
})).rejects.toThrow("rollback owner");
|
||||
await settle(); expect(await publications()).toEqual([]); await quiet();
|
||||
await db.transaction(async tx => {
|
||||
await append(tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await publications()).toEqual([]);
|
||||
});
|
||||
await settle(); expect(await publications()).toHaveLength(1); await quiet();
|
||||
const signals: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind));
|
||||
try { expect((await append(db)).disposition).toBe("duplicate"); expect(signals).toEqual([]); }
|
||||
finally { unsubscribe(); }
|
||||
await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it.each(["issue", "run"] as const)("disarms after terminalization wins during a contended %s", async owner => {
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
let release!: () => void, entered!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(resolve => { entered = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
if (owner === "issue") await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
else await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
|
||||
entered(); await hold;
|
||||
await tx.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, runId));
|
||||
});
|
||||
try {
|
||||
await ready; await start(); expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
release(); await lock; await settle(); expect(await publications()).toEqual([]); await quiet();
|
||||
expect(errors).toEqual([]);
|
||||
} finally { release(); await lock; }
|
||||
});
|
||||
|
||||
it.each(["first write", "startup"] as const)("retains %s through drain without passive SQL", async kind => {
|
||||
if (kind === "first write") { await start(); await settle(); }
|
||||
canRun = false;
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
if (kind === "startup") await start();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
|
||||
expect(await publications()).toEqual([]); await quiet();
|
||||
canRun = true; await vi.advanceTimersByTimeAsync(30_000); await settle();
|
||||
expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("retains failed startup selection through drain and retries after resume", async () => {
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("temporary read failure"); });
|
||||
try { await start(); } finally { select.mockRestore(); }
|
||||
expect(errors.splice(0)).toHaveLength(1); canRun = false;
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
|
||||
await quiet(); expect(await publications()).toEqual([]);
|
||||
canRun = true; await vi.advanceTimersByTimeAsync(30_000); await settle();
|
||||
expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("leaves unrelated direct log events on their atomic statement path without projection intent", async () => {
|
||||
await start(); await settle();
|
||||
const signals: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind));
|
||||
const transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await appendHeartbeatRunEvent(db, event(false, "item.delta"));
|
||||
expect(transaction).not.toHaveBeenCalled(); expect(signals).toEqual([]);
|
||||
} finally { transaction.mockRestore(); unsubscribe(); }
|
||||
await quiet(); expect(await publications()).toEqual([]); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("wakes saved trace reprojection without live runner ingress and leaves duplicate recovery quiet", async () => {
|
||||
await start(); await settle(); await quiet();
|
||||
const projection = projectCodexWorkspaceDiffsFromTrace([{
|
||||
kind: "frame", direction: "provider_to_client", frameId: 1,
|
||||
rawBase64: Buffer.from(JSON.stringify({ method: "turn/diff/updated", params: {
|
||||
turnId: "turn-1", diff: "diff --git a/private.txt b/private.txt\n--- a/private.txt\n+++ b/private.txt\n@@ -1 +1 @@\n-private\n+PRIVATE replacement\n",
|
||||
} })).toString("base64"),
|
||||
}]);
|
||||
expect(projection.turns).toHaveLength(1);
|
||||
const input = { traceId: randomUUID(), companyId: f.companyId, agentId: f.agentId, runId, projection };
|
||||
expect(await persistReprojectedWorkspaceDiffs(db, input)).toMatchObject({ created: 1 });
|
||||
await settle();
|
||||
expect(await publications()).toEqual([expect.objectContaining({ payload: expect.objectContaining({ text: "Maya is working with files…" }) })]);
|
||||
await quiet();
|
||||
const signals: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind));
|
||||
try { expect(await persistReprojectedWorkspaceDiffs(db, input)).toMatchObject({ created: 0 }); expect(signals).toEqual([]); }
|
||||
finally { unsubscribe(); }
|
||||
await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("retains a committed event that arrives before the current projection pass returns", async () => {
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const secondRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: secondRunId, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId,
|
||||
runtimeMode: "native", status: "running", responsibleUserId: f.userId, contextSnapshot: source.contextSnapshot,
|
||||
});
|
||||
await db.insert(chatPublications).values({
|
||||
companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId,
|
||||
issueId: f.issueId, idempotencyKey: `run:${secondRunId}:working:${f.endpointId}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" }, state: "published",
|
||||
createdAt: new Date(Date.now() - 60_000),
|
||||
});
|
||||
await start(); await settle();
|
||||
let held = false, release!: () => void;
|
||||
const pending = new Promise<void>(resolve => { release = resolve; });
|
||||
const transaction = db.transaction.bind(db);
|
||||
const barrier = vi.spyOn(db, "transaction").mockImplementation(async (callback, options) => {
|
||||
const result = await transaction(callback, options);
|
||||
// Pause after the first projection commits, before its pass can disarm.
|
||||
// Producer transactions return event records, not an inserted row count.
|
||||
if (result === 1 && !held) { held = true; await pending; }
|
||||
return result;
|
||||
});
|
||||
try {
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
await vi.waitFor(() => expect(held).toBe(true), { timeout: 5000 });
|
||||
await appendHeartbeatRunEvent(db, { ...event(false), runId: secondRunId });
|
||||
release(); await settle();
|
||||
expect(await publications()).toHaveLength(2); barrier.mockRestore();
|
||||
await quiet(); expect(errors).toEqual([]);
|
||||
} finally { release(); barrier.mockRestore(); }
|
||||
});
|
||||
|
||||
it("limits an event wake to its issue and reconstructs other saved scopes on restart", async () => {
|
||||
await start(); await settle();
|
||||
// Seed rows without signals to expose an accidentally global event wake.
|
||||
const others = [await fixtures.create(f.companyId), await fixtures.create()];
|
||||
for (const other of others) {
|
||||
const id = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id, companyId: other.companyId, agentId: other.agentId, nativeIssueId: other.issueId,
|
||||
runtimeMode: "native", status: "running", responsibleUserId: other.userId,
|
||||
contextSnapshot: { issueId: other.issueId, source: "chat:slack", endpointId: other.endpointId, wakeCommentId: other.commentId },
|
||||
});
|
||||
await db.insert(chatPublications).values({
|
||||
companyId: other.companyId, endpointId: other.endpointId, conversationId: other.conversationId,
|
||||
issueId: other.issueId, idempotencyKey: `run:${id}:working:${other.endpointId}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" }, state: "published",
|
||||
createdAt: new Date(Date.now() - 60_000),
|
||||
});
|
||||
await db.insert(heartbeatRunEvents).values({ companyId: other.companyId, agentId: other.agentId,
|
||||
runId: id, seq: 1, eventType: "item.completed" });
|
||||
}
|
||||
const saved = (issueId: string) => db.select().from(chatPublications).where(and(
|
||||
eq(chatPublications.issueId, issueId), like(chatPublications.idempotencyKey, "%:native:%"),
|
||||
));
|
||||
await appendHeartbeatRunEvent(db, event(true)); await settle();
|
||||
expect(await publications()).toHaveLength(1);
|
||||
for (const other of others) expect(await saved(other.issueId)).toEqual([]);
|
||||
await quiet(); await coordinator!.stop(); await start(); await settle();
|
||||
for (const other of others) expect(await saved(other.issueId)).toHaveLength(1);
|
||||
await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("drains more than one issue batch and becomes quiet with all progress retained", async () => {
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const ids = Array.from({ length: 105 }, () => randomUUID());
|
||||
await db.insert(heartbeatRuns).values(ids.map(id => ({
|
||||
id, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId,
|
||||
runtimeMode: "native", status: "running", responsibleUserId: f.userId, contextSnapshot: source.contextSnapshot,
|
||||
})));
|
||||
await db.insert(chatPublications).values(ids.map(id => ({
|
||||
companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId,
|
||||
issueId: f.issueId, idempotencyKey: `run:${id}:working:${f.endpointId}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" }, state: "published",
|
||||
createdAt: new Date(Date.now() - 60_000),
|
||||
})));
|
||||
await db.insert(heartbeatRunEvents).values(ids.map(id => ({
|
||||
companyId: f.companyId, agentId: f.agentId, runId: id, seq: 1, eventType: "item.completed",
|
||||
})));
|
||||
await start(); await settle();
|
||||
expect(await publications()).toHaveLength(ids.length); await quiet(); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
it("unsubscribes at shutdown and reconstructs a later event on restart", async () => {
|
||||
await start(); await settle(); await coordinator!.stop();
|
||||
await appendHeartbeatRunEvent(db, event(true));
|
||||
await quiet(); expect(await publications()).toEqual([]);
|
||||
await start(); await settle(); expect(await publications()).toHaveLength(1); expect(errors).toEqual([]);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -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, projectNativeProgress: async () => 0,
|
||||
processPendingDeliveries: async () => undefined,
|
||||
flushPublications: async () => undefined, onError,
|
||||
});
|
||||
try {
|
||||
|
||||
+4
-22
@@ -195,7 +195,7 @@ 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 { enqueueSafeNativeChatProgress } from "./services/chat-run-publications.js";
|
||||
import { nativeChatProgressWork } from "./services/chat-native-progress-work.js";
|
||||
import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js";
|
||||
import {
|
||||
createCoalescedAsyncTrigger,
|
||||
@@ -324,7 +324,6 @@ export function shouldEnablePrivateHostnameGuard(opts: {
|
||||
type ChatReconciliationLane =
|
||||
| "deliveries"
|
||||
| "GitHub webhook recovery"
|
||||
| "native progress"
|
||||
| "publications"
|
||||
| "Slack file receipts"
|
||||
| "Slack session status";
|
||||
@@ -337,7 +336,6 @@ type ChatReconciliationLane =
|
||||
export function createChatReconciliationCoordinator(input: {
|
||||
processPendingDeliveries: () => Promise<unknown>;
|
||||
processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
|
||||
projectNativeProgress: () => Promise<number>;
|
||||
flushPublications: () => Promise<unknown>;
|
||||
processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
|
||||
processPendingSlackSessionSyncs?: () => Promise<unknown>;
|
||||
@@ -349,16 +347,6 @@ export function createChatReconciliationCoordinator(input: {
|
||||
run: input.flushPublications,
|
||||
onError: (error) => input.onError("publications", error),
|
||||
});
|
||||
const progressReconciliation = createCoalescedAsyncTrigger({
|
||||
run: async () => {
|
||||
const inserted = await input.projectNativeProgress();
|
||||
// Existing final/question publications never wait on this optional
|
||||
// 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("native progress", error),
|
||||
});
|
||||
const start = (
|
||||
lane: ChatReconciliationLane,
|
||||
task: () => Promise<unknown>,
|
||||
@@ -383,27 +371,21 @@ export function createChatReconciliationCoordinator(input: {
|
||||
input.processFailedGitHubWebhookDeliveries,
|
||||
);
|
||||
}
|
||||
progressReconciliation.poll();
|
||||
publicationReconciliation.poll();
|
||||
if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
|
||||
if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs);
|
||||
},
|
||||
notifyPublications() {
|
||||
progressReconciliation.notify();
|
||||
publicationReconciliation.notify();
|
||||
},
|
||||
stop() {
|
||||
stopped = true;
|
||||
progressReconciliation.stop();
|
||||
publicationReconciliation.stop();
|
||||
},
|
||||
async drain() {
|
||||
await Promise.allSettled([
|
||||
...inFlight.values(),
|
||||
progressReconciliation.drain(),
|
||||
]);
|
||||
// Projecting the final batch can notify dispatch after an earlier drain
|
||||
// would have returned. Join dispatch only after its producer has drained.
|
||||
await publicationReconciliation.drain();
|
||||
},
|
||||
};
|
||||
@@ -1305,6 +1287,7 @@ export async function createApp(
|
||||
chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatRunMilestoneWork(db, opts.authPublicBaseUrl).register(
|
||||
deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(),
|
||||
);
|
||||
@@ -1318,7 +1301,6 @@ export async function createApp(
|
||||
processPendingDeliveries: async () => undefined,
|
||||
processFailedGitHubWebhookDeliveries: () =>
|
||||
chatChannels.processFailedGitHubWebhookDeliveries(),
|
||||
projectNativeProgress: () => enqueueSafeNativeChatProgress(db),
|
||||
flushPublications: () => reconcileChatPublicationMaintenance(),
|
||||
onError: (lane, err) => {
|
||||
logger.error({ err, lane }, `Failed to reconcile chat ${lane}`);
|
||||
@@ -1330,8 +1312,8 @@ export async function createApp(
|
||||
chatReconciliation.notifyPublications();
|
||||
},
|
||||
);
|
||||
// Native progress and remaining publication maintenance still use this cadence.
|
||||
// Lifecycle milestones use committed changes and their own recovery worker.
|
||||
// Remaining publication maintenance still uses this cadence. Lifecycle and
|
||||
// native progress projection have independent commit/deadline workers.
|
||||
// Inbox, outbox, and file receipt queues have independent commit/deadline wakes.
|
||||
let chatPublicationTimer: ReturnType<typeof setInterval> | null = setInterval(
|
||||
() => {
|
||||
|
||||
@@ -113,7 +113,7 @@ export function createChatPublicationResolution(db: Db, options: ResolutionOptio
|
||||
code: "chat_file_transfer_resolution_required",
|
||||
},
|
||||
);
|
||||
await notifyChatPublicationWork(tx);
|
||||
await notifyChatPublicationWork(tx, current);
|
||||
return tx
|
||||
.update(chatPublications)
|
||||
.set({
|
||||
@@ -218,7 +218,7 @@ export function createChatPublicationResolution(db: Db, options: ResolutionOptio
|
||||
action === "retry_anyway"
|
||||
) { await recordPhotonPublicationRetry(tx, endpointId, publication, userId); }
|
||||
const now = new Date();
|
||||
await notifyChatPublicationWork(tx, action === "retry_anyway" ? undefined : publication);
|
||||
await notifyChatPublicationWork(tx, publication);
|
||||
await notifyVoicePublication(tx, initialRecord.endpoint);
|
||||
await tx
|
||||
.update(chatPublications)
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import type { ChatRunProjectionScope } from "./chat-run-publication-policy.js";
|
||||
|
||||
export function notifyChatNativeProgress(tx: object, scope: ChatRunProjectionScope) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatNativeProgress, JSON.stringify(scope));
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
import type { ChatRunProjectionScope } from "./chat-run-publication-policy.js";
|
||||
import { loadNativeChatProgressScopes, reconcileNativeChatProgress } from "./chat-native-progress.js";
|
||||
|
||||
type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
|
||||
const batchSize = 100;
|
||||
|
||||
export function nativeChatProgressWork(db: Db) {
|
||||
async function load(scope?: ChatRunProjectionScope, signal?: AbortSignal): Promise<Entry[]> {
|
||||
return (await loadNativeChatProgressScopes(db, scope, signal)).map(row => ({
|
||||
...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0,
|
||||
}));
|
||||
}
|
||||
async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
|
||||
const result = await reconcileNativeChatProgress(db, {
|
||||
scope: { companyId: entry.companyId, issueId: entry.issueId }, limit: batchSize, signal,
|
||||
});
|
||||
if (result.inserted === batchSize) return { ...entry, dueAt: Date.now() };
|
||||
return result.contended ? { ...entry, dueAt: Date.now() + 1000 } : null;
|
||||
}
|
||||
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.chatNativeProgress,
|
||||
decodeScope: (hint: string): ChatRunProjectionScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: Entry, scope?: ChatRunProjectionScope) => !scope ||
|
||||
(entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)),
|
||||
load, check,
|
||||
}, canRun);
|
||||
}
|
||||
return { load, check, register };
|
||||
}
|
||||
@@ -7,7 +7,7 @@ import { projectSafeChatPublication } from "./chat-publication-projection.js";
|
||||
import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js";
|
||||
import { resolveChatOriginPublicationBindings } from "./issues.js";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js";
|
||||
import { allowsAutomaticRunPublication, chatRunProjectionWindow, type ChatRunProjectionScope } from "./chat-run-publication-policy.js";
|
||||
import { SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, safeNativeChatProgressForEvent } from "./safe-native-chat-progress.js";
|
||||
|
||||
type SafeNativeChatProgressCandidate = {
|
||||
@@ -26,12 +26,10 @@ type SafeNativeChatProgressCandidate = {
|
||||
|
||||
const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000;
|
||||
|
||||
/** Only closed event types become coarse progress; event payloads stay local. */
|
||||
export async function enqueueSafeNativeChatProgress(
|
||||
db: Db,
|
||||
options: { since?: Date; limit?: number } = {},
|
||||
): Promise<number> {
|
||||
const input = chatRunProjectionWindow(options);
|
||||
type NativeProgressCursor = Pick<SafeNativeChatProgressCandidate, "eventCreatedAt" | "runId" | "conversationId">;
|
||||
type NativeProgressInput = { since?: Date; limit?: number; scope?: ChatRunProjectionScope; signal?: AbortSignal };
|
||||
|
||||
function nativeProgressPages(db: Db, since: Date, scope?: ChatRunProjectionScope) {
|
||||
const basePublication = alias(
|
||||
chatPublications,
|
||||
"safe_native_progress_base_publication",
|
||||
@@ -41,17 +39,7 @@ export async function enqueueSafeNativeChatProgress(
|
||||
"safe_native_progress_later_event",
|
||||
);
|
||||
const issueIdFromContext = boundRunIssueIdSql();
|
||||
let inserted = 0;
|
||||
let cursor: {
|
||||
eventCreatedAt: string;
|
||||
runId: string;
|
||||
conversationId: string;
|
||||
} | null = null;
|
||||
|
||||
while (inserted < input.limit) {
|
||||
const pageSize = Math.max(25, Math.min(200, input.limit - inserted));
|
||||
const pageCursor: typeof cursor = cursor;
|
||||
const rows: SafeNativeChatProgressCandidate[] = await db
|
||||
return async (pageSize: number, pageCursor: NativeProgressCursor | null): Promise<SafeNativeChatProgressCandidate[]> => await db
|
||||
.select({
|
||||
eventId: heartbeatRunEvents.id,
|
||||
eventSeq: heartbeatRunEvents.seq,
|
||||
@@ -118,7 +106,9 @@ export async function enqueueSafeNativeChatProgress(
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
gte(heartbeatRunEvents.createdAt, input.since),
|
||||
gte(heartbeatRunEvents.createdAt, since),
|
||||
scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
|
||||
scope?.issueId ? eq(chatConversations.issueId, scope.issueId) : undefined,
|
||||
sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`,
|
||||
notExists(
|
||||
db
|
||||
@@ -147,6 +137,31 @@ export async function enqueueSafeNativeChatProgress(
|
||||
asc(chatConversations.id),
|
||||
)
|
||||
.limit(pageSize);
|
||||
}
|
||||
|
||||
export async function loadNativeChatProgressScopes(db: Db, scope?: ChatRunProjectionScope, signal?: AbortSignal) {
|
||||
const pages = nativeProgressPages(db, chatRunProjectionWindow({}).since, scope);
|
||||
const scopes = new Map<string, { companyId: string; issueId: string }>();
|
||||
let cursor: NativeProgressCursor | null = null;
|
||||
for (;;) {
|
||||
signal?.throwIfAborted();
|
||||
const rows = await pages(200, cursor);
|
||||
for (const row of rows) scopes.set(`${row.companyId}:${row.issueId}`, { companyId: row.companyId, issueId: row.issueId });
|
||||
if (rows.length < 200) return [...scopes.values()];
|
||||
cursor = rows[rows.length - 1];
|
||||
}
|
||||
}
|
||||
|
||||
/** A suppressed event is settled; lock contention still owes another attempt. */
|
||||
export async function reconcileNativeChatProgress(db: Db, options: NativeProgressInput = {}) {
|
||||
const input = chatRunProjectionWindow(options);
|
||||
const pages = nativeProgressPages(db, input.since, options.scope);
|
||||
let inserted = 0, contended = false;
|
||||
let cursor: NativeProgressCursor | null = null;
|
||||
while (inserted < input.limit) {
|
||||
options.signal?.throwIfAborted();
|
||||
const pageSize = Math.max(25, Math.min(200, input.limit - inserted));
|
||||
const rows = await pages(pageSize, cursor);
|
||||
if (rows.length === 0) break;
|
||||
const lastRow = rows.at(-1)!;
|
||||
cursor = {
|
||||
@@ -156,6 +171,7 @@ export async function enqueueSafeNativeChatProgress(
|
||||
};
|
||||
|
||||
for (const row of rows) {
|
||||
options.signal?.throwIfAborted();
|
||||
if (inserted >= input.limit) break;
|
||||
const result = await db.transaction(async (tx) => {
|
||||
// Keep the normal issue -> run order used by native finalization. The
|
||||
@@ -174,7 +190,12 @@ export async function enqueueSafeNativeChatProgress(
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!currentIssue) return 0;
|
||||
if (!currentIssue) {
|
||||
const [existing] = await tx.select({ id: issues.id }).from(issues)
|
||||
.where(and(eq(issues.id, row.issueId), eq(issues.companyId, row.companyId)));
|
||||
contended ||= Boolean(existing);
|
||||
return 0;
|
||||
}
|
||||
// Native event admission serializes on the run row. Taking the same
|
||||
// short lock makes terminal status, a later event, or a final comment
|
||||
// win before this non-terminal projection can be inserted.
|
||||
@@ -192,7 +213,15 @@ export async function enqueueSafeNativeChatProgress(
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!currentRun) return 0;
|
||||
if (!currentRun) {
|
||||
const [existing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, row.runId), eq(heartbeatRuns.companyId, row.companyId),
|
||||
eq(heartbeatRuns.agentId, row.agentId), eq(heartbeatRuns.runtimeMode, "native"),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
));
|
||||
contended ||= Boolean(existing);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const destination = await tx
|
||||
.select({ id: chatConversations.id })
|
||||
@@ -389,5 +418,10 @@ export async function enqueueSafeNativeChatProgress(
|
||||
}
|
||||
if (rows.length < pageSize) break;
|
||||
}
|
||||
return inserted;
|
||||
return { inserted, contended };
|
||||
}
|
||||
|
||||
/** Explicit/manual callers retain the original count-only contract. */
|
||||
export async function enqueueSafeNativeChatProgress(db: Db, options: NativeProgressInput = {}): Promise<number> {
|
||||
return (await reconcileNativeChatProgress(db, options)).inserted;
|
||||
}
|
||||
@@ -1,9 +1,13 @@
|
||||
import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
/** Lineage changes can authorize descendants, so invalidate the issue rather
|
||||
* than just the run whose row changed. Configuration may invalidate a company. */
|
||||
export type ChatRunMilestoneScope = { companyId: string; issueId?: string };
|
||||
|
||||
export function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
|
||||
export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
|
||||
// Lifecycle and coarse native progress share lineage and authority. Extend
|
||||
// this eligibility boundary once rather than adding wrappers to its callers.
|
||||
await notifyChatNativeProgress(tx, scope);
|
||||
}
|
||||
@@ -17,3 +17,5 @@ export function chatRunProjectionWindow(input: { since?: Date; limit?: number })
|
||||
limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)),
|
||||
};
|
||||
}
|
||||
|
||||
export type ChatRunProjectionScope = { companyId: string; issueId?: string };
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
|
||||
import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js";
|
||||
import { enqueueSafeNativeChatProgress } from "./chat-native-progress.js";
|
||||
export { enqueueSafeNativeChatProgress } from "./chat-native-progress.js";
|
||||
@@ -646,6 +647,9 @@ export async function enqueueChatRunLifecycleMilestones(
|
||||
if (explicitlyAuthoredPublication.length > 0) continue;
|
||||
}
|
||||
const result = await db.transaction(async (tx) => {
|
||||
if (milestone === "working") {
|
||||
await notifyChatNativeProgress(tx, { companyId: row.companyId, issueId: row.issueId });
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.insert(chatPublications)
|
||||
|
||||
@@ -54,6 +54,7 @@ export const DELIVERY_QUEUES = {
|
||||
chatVerification: "chat-verification",
|
||||
chatInbound: "chat-inbound",
|
||||
chatPublications: "chat-publications",
|
||||
chatNativeProgress: "chat-native-progress",
|
||||
chatRunMilestones: "chat-run-milestones",
|
||||
chatReceipts: "chat-slack-receipts",
|
||||
browser: "browser-use",
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js";
|
||||
import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { and, asc, eq, getTableColumns, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
@@ -68,7 +71,8 @@ export async function appendHeartbeatRunEvent(
|
||||
db: Db,
|
||||
input: AppendHeartbeatRunEventInput,
|
||||
): Promise<AppendHeartbeatRunEventResult> {
|
||||
if (!input.nativeSource && !input.retryExhaustion) {
|
||||
const projectsProgress = isSafeNativeChatProgressEventType(input.eventType);
|
||||
if (!input.nativeSource && !input.retryExhaustion && !projectsProgress) {
|
||||
const allocated = db.$with("allocated_run_event", {}).as(sql`
|
||||
update ${heartbeatRuns}
|
||||
set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,
|
||||
@@ -94,7 +98,8 @@ export async function appendHeartbeatRunEvent(
|
||||
}
|
||||
return db.transaction(async (tx) => {
|
||||
const run = await tx
|
||||
.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId })
|
||||
.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
runtimeMode: heartbeatRuns.runtimeMode, issueId: boundRunIssueIdSql() })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.for("update")
|
||||
@@ -165,6 +170,9 @@ export async function appendHeartbeatRunEvent(
|
||||
}
|
||||
}
|
||||
|
||||
if (projectsProgress && run.runtimeMode === "native" && run.issueId) {
|
||||
await notifyChatNativeProgress(tx, { companyId: run.companyId, issueId: run.issueId });
|
||||
}
|
||||
const seq = await allocateHeartbeatRunEventSeq(
|
||||
tx as unknown as Db,
|
||||
input.runId,
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js";
|
||||
import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
import { and, desc, eq, inArray, lt } from "drizzle-orm";
|
||||
@@ -290,6 +292,9 @@ export class NativeRunCoordinatorStore {
|
||||
throw new Error("native_event_source_gap");
|
||||
}
|
||||
|
||||
if (isSafeNativeChatProgressEventType(event.eventType)) {
|
||||
await notifyChatNativeProgress(tx, { companyId: this.#binding.companyId, issueId: this.#binding.issueId });
|
||||
}
|
||||
const cursor = run.nextEventSeq;
|
||||
const [inserted] = await tx
|
||||
.insert(heartbeatRunEvents)
|
||||
|
||||
Reference in new issue
Block a user