mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 04:25:52 +01:00
fix(server): wake native chat progress from committed changes (#17639)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat users receive coarse progress updates while native runs execute. > - A recurring chat pass scans saved run events even when no progress changes. > - This traffic prevents an idle database from staying idle. > - Events, lineage changes, and publication state already identify when projection can change. > - This PR gives native progress its own commit-driven worker and schedules only unfinished work. > - Idle progress stops scanning while restart and recovery preserve delivery. ## Linked Issues or Issue Description Refs #17606. This is slice 29 of the recurring-server-work stack, based on `codex/sleep-28-chat-run-milestones`. It owns native chat progress only. Related historical optimization: #16845. The parent and this PR must land in stack order. **What happened?** Native chat progress is still called by the one-second chat reconciliation loop. It repeatedly selects run events with no new work. Its cursor also truncates PostgreSQL timestamp precision, which can repeat a rejected full page forever. **Expected behavior** Committed events and eligibility changes should wake the projection owner. Existing rows should reconstruct work after restart. A settled worker should have no recurring database activity. Full batches, lock contention, and errors must retain recovery. **Steps to reproduce** Start an otherwise idle instance with chat history. Observe the native progress lane in the recurring chat coordinator. For the cursor defect, create a full page of suppressed candidates sharing a microsecond timestamp and one eligible later candidate. The old cursor fails to advance. ## What Changed - Extract native progress from lifecycle milestones into its owning service. Keep the public combined API and budget compatible. - Register a scoped worker with the existing delivery coordinator and scheduler. Remove native progress from the recurring chat coordinator. - Notify before allowlisted event writes in both persistence owners. Preserve outer commit, rollback, duplicate detection, and the ordinary non-progress log fast path. - Extend the existing lineage/authority notification boundary once. Failed-publication replay and operator retry carry their issue scope. First working publications signal progress so commit ordering cannot strand an already-inspected event. - Limit notification payloads to company/task IDs so wider caller inputs never serialize database handles or private fields. Load task-binding fields only for progress event kinds; other native/recovery events retain their narrow lock query. - Preserve current safe phrases, cadence, phase deduplication, provider policy, authorization, and final-output precedence. - Retry concrete issue/run lock contention. Disarm deleted or terminal candidates. Preserve scoped dirty work across batches, failures, drain and shutdown. - Preserve PostgreSQL timestamp precision in pagination. Add a bounded regression for the former infinite rejected-page scan. - Document the producer audit, lifecycle contract, failures, and remaining limits in `doc/plans/2026-10-25-native-chat-progress-sleep-audit.md`. ## Verification Published head: `b7ee820e49fd06c780b36b79d9b1c0883557fb73`, targeting master after #17606 merged at `5f0da562dbfdf3229f3cd4bb3dc742ab541986b4`. A normal ancestry merge preserves the complete 18-path native-progress patch byte-for-byte relative to its verified parent, with all six unrelated current master paths intact. No new production change was introduced during final reconciliation. - **163 distinct selected cases pass on the verified unchanged patch:** 137 across eight complete suites (native progress, atomic event append, all five milestone worker suites, and app admission wiring), plus 26 real chat integration cases selected by `native progress`. The other 1,595 chat integration cases were skipped, not passed. - The full worker suites cover restart, committed events, trace reprojection, duplicate/no-progress events, first-working publication ordering, concrete lock contention, issue scope, backlog, drain/resume, failures and shutdown. Settled history leaves no recurring SQL. The real chat cases verify operator retry and replay, microsecond pagination, authority, cadence, phase deduplication, final/question precedence and redaction. - The sole earlier parent conflict preserves both this slice's issue-scoped retry wake and the exact voice conversation scope. The current parent's milestone producers, onboarding completion guard, identity query/read-scope coverage, migration 0397 and voice cleanup/retention repairs remain present. - Server production typechecking, build and module-boundary checks pass. The selected server tests live under the excluded `src/__tests__` path and were executed separately. Each database owner uses a new isolated app-migrated PostgreSQL fixture. - [Current-parent acceptance and failure ledger](https://github.com/paperclipai/paperclip/blob/b7ee820e49fd06c780b36b79d9b1c0883557fb73/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md#current-milestone-parent-acceptance-october-28) records current counts and scope. Older combined checks and failures retain their original revision scope and are not added to 163. No new table, dependency, workflow, migration, or scheduler contract is introduced. - **Exact-head Greptile is 5/5 and all 55 current checks are clean.** No Ponytail review is present, which is an explicitly authorized nonblocker; no failing verdict is waived. Older head/parent reviews are not current passes. Actual failing Ponytail findings must be fixed. Missing/stopped/paused turns without a verdict are authorized nonblockers; active unfinished reviews retain the original per-head timer with a 25-minute monitor cutoff respecting the 30-minute maximum. - No complete local repository matrix, live provider tenant, external wake, physical host/database suspension or savings claim is made. Following stack duties remain separate. ## Risks - Signals are process-local under the existing single active writer contract. Independent SQL writers or another active process need an explicit wake integration. - Allowlisted direct log writes now use a transaction. Native bound runs register transaction intent; unrelated direct event types retain their atomic-statement path. - The shared eligibility boundary wakes both lifecycle and native projection. It does not decide provider policy or copy event payloads. - The existing 24-hour selection window, event-time cadence and authority checks remain. A suppressed event is not retried just because wall-clock time passes. - This establishes worker quiet, not whole-instance suspension. Cloud alarm coverage and conservative sleep gates do not change. Other chat maintenance remains on its existing cadence. - This actual-master reconciliation still requires fresh exact-head review and CI before protected merge. Subsequent slices retain their own gates. ## Model Used OpenAI Codex, GPT-6 family, with repository inspection, code execution and browser-test tools. The exact serving variant and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
0791da0cc5
commit
1ed0af4186
18 files changed
+1282
-566
No files matched your search
@@ -2170,3 +2170,21 @@ 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
|
||||
|
||||
PR [#17639](https://github.com/paperclipai/paperclip/pull/17639), 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. A second fresh browser run passes all
|
||||
16 cases at `bcf2f38ff5`; final typecheck passes too. Remote checks/review 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.
|
||||
@@ -0,0 +1,225 @@
|
||||
# Native chat progress: sleep migration audit
|
||||
|
||||
Parent: #17606 at `ed76256b979541b962fa784016a550c32920d670`.
|
||||
Branch: `codex/sleep-29-native-chat-progress`. PR: #17639.
|
||||
Status: implementation, focused PostgreSQL acceptance, and browser acceptance
|
||||
complete; exact-head remote checks and review pending. This is not a deployed sleep claim.
|
||||
|
||||
## Duty and owner
|
||||
|
||||
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, authorization and insertion.
|
||||
`safe-native-chat-progress.ts` owns the normalized allowlist and phrase policy.
|
||||
`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.
|
||||
|
||||
## Durable contract and triggers
|
||||
|
||||
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.
|
||||
|
||||
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 two event writers register awaited intent before insertion, after duplicate
|
||||
and ownership checks, inside their existing transaction boundaries:
|
||||
|
||||
- `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.
|
||||
|
||||
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.
|
||||
|
||||
## Scheduling, contention and lifetime
|
||||
|
||||
`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.
|
||||
|
||||
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.
|
||||
|
||||
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.
|
||||
|
||||
## Evidence and failures retained
|
||||
|
||||
- `/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. `/tmp/sleep29-final-e2e.log` then
|
||||
passes all 16 cases at committed implementation `bcf2f38ff5` with a second
|
||||
fresh isolated database. `/tmp/sleep29-final-typecheck.log` also passes with
|
||||
the final hook. No retries were needed in either browser run.
|
||||
|
||||
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.
|
||||
|
||||
## Limits and remaining acceptance
|
||||
|
||||
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. Exact-head CI and review remain required before this slice is ready. The whole stack remains unmerged.
|
||||
|
||||
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 interval remains. Preserve that newer contract
|
||||
when migrating its separate retention owner.
|
||||
|
||||
## CI repair: narrow notification payload (2026-10-25)
|
||||
|
||||
CI at `34df8b0d01` failed native status/finalization consumers across server,
|
||||
chat, Runner, and serialized suites. Local reproduction of the existing
|
||||
same-run Slack response-recovery test failed before the repair. Its persisted
|
||||
`native_run_finalizations.failure_detail.message` identified circular JSON
|
||||
serialization of a Drizzle table through the caller's database handle.
|
||||
|
||||
The new native-progress notifier serialized the entire structurally compatible
|
||||
scope argument. Existing lifecycle callers legitimately pass wider domain inputs
|
||||
(including `db`) through that typed boundary. Serialize only `companyId` and
|
||||
`issueId`, as the lifecycle notifier already does; do not constrain every caller
|
||||
or serialize domain/private fields into work hints.
|
||||
|
||||
After repair, 15 existing chat finalization/response-recovery/authority cases
|
||||
passed against external PostgreSQL, all 32 native-progress worker/startup cases
|
||||
passed, and server typecheck passed. The attempted broader native-finalization
|
||||
and session-resumption selection failed during local embedded PostgreSQL
|
||||
initialization (bootstrap exit 1), before tests could run; it is not acceptance
|
||||
evidence. Fresh CI must verify those native consumers. Before/after logs:
|
||||
`/tmp/sleep29-finalization-repro.log`, `/tmp/sleep29-finalization-fixed.log`,
|
||||
`/tmp/sleep29-notification-fix-workers.log`,
|
||||
`/tmp/sleep29-notification-fix-typecheck.log`, and
|
||||
`/tmp/sleep29-native-finalization-acceptance.log`.
|
||||
|
||||
## CI repair: keep non-progress event admission narrow (2026-10-25)
|
||||
|
||||
CI at `3447973f6f` clears the prior finalization failures. Its sole failing test
|
||||
file is `heartbeat-run-event-atomic-append.test.ts`: native lifecycle and retry
|
||||
exhaustion events unnecessarily select runtime mode and the task-binding SQL,
|
||||
which can read a large context snapshot. Those events do not project progress.
|
||||
The lock query now selects the extra binding only for allowlisted progress
|
||||
kinds. Keep the existing strict SQL/payload regression assertions unchanged.
|
||||
|
||||
The atomic-event suite now supports the repository's guarded external test DB
|
||||
convention, allowing its actual PostgreSQL SQL assertions to run locally despite
|
||||
embedded bootstrap failure. The original implementation reproduces both CI
|
||||
failures in `/tmp/sleep29-atomic-events-before.log`. That first external run also
|
||||
exposed fixture cleanup ordering; cleanup now removes event/run/agent rows before
|
||||
companies and always closes its pools. The repaired atomic-event and progress
|
||||
worker suites pass all 25 cases in `/tmp/sleep29-atomic-progress-fixed.log`;
|
||||
server typecheck passes in `/tmp/sleep29-atomic-progress-typecheck.log`.
|
||||
|
||||
Greptile reviewed `3447973f6f` at 5/5 with no threads. A new full CI/review pass
|
||||
is required for this second repair; do not carry forward readiness from its
|
||||
parent head.
|
||||
|
||||
|
||||
## Current milestone-parent acceptance (October 28)
|
||||
|
||||
The existing native-progress slice is prepared with a normal merge of the verified
|
||||
local milestone parent `eef52b9dd759417bd7a2fea835357fbfeb460111`. Its previous
|
||||
published head is `08d07ead105dc89c2af9b62f4d90bff8cc93540c`, with original own
|
||||
base `7991c2fa2ce7298bee69b25dbc6cb60a5e2fe250`. All unrelated parent changes,
|
||||
including the current migration 0397 and authority, voice, and identity repairs,
|
||||
remain present. The only explicit-base conflict was publication resolution: keep
|
||||
this slice's issue-scoped retry notification and the parent's exact voice
|
||||
conversation scope. No new duty, table, dependency, or scheduler is introduced.
|
||||
|
||||
Fresh separate PostgreSQL fixtures pass **163 distinct selected cases**:
|
||||
**137 across eight complete suites** (native progress, atomic event append,
|
||||
all five milestone worker suites, and app admission wiring), plus **26 real chat
|
||||
integration cases** selected by `native progress`. These include replay/operator
|
||||
retry, microsecond pagination, lock contention, current authority, cadence, phase
|
||||
coalescing, final/question precedence, and redaction. The other **1,595 chat cases
|
||||
were skipped**, not passed. Server production typechecking, build, and module
|
||||
boundaries also pass; test files were executed rather than statically checked.
|
||||
|
||||
Commands, separate fixture names, output, and zero exit-status receipts are in
|
||||
`/tmp/sleep29-current-parent-acceptance-command.json` and
|
||||
`/tmp/sleep29-current-parent-{acceptance,producers,types,build,boundaries}.log`.
|
||||
Earlier verification and failures above keep their stated revision scope. This
|
||||
is local preparation only: final landed-parent reconciliation, all fresh exact-
|
||||
head remote gates, and Greptile 5/5 remain required before merge. This does not
|
||||
qualify a live provider, external alarm, whole-instance suspension, or savings.
|
||||
@@ -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";
|
||||
@@ -53367,6 +53368,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 {
|
||||
@@ -65646,6 +65698,91 @@ 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 {
|
||||
const valid = await context.createRun("eligible after rejected page");
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, valid.runId));
|
||||
const rejectedIds = Array.from({ length: 30 }, () => randomUUID());
|
||||
await db.insert(heartbeatRuns).values(rejectedIds.map(id => ({
|
||||
id, companyId: source.companyId, agentId: source.agentId,
|
||||
runtimeMode: "native", status: "running",
|
||||
contextSnapshot: { ...source.contextSnapshot, endpointId: randomUUID() },
|
||||
})));
|
||||
await db.insert(chatPublications).values(rejectedIds.map(id => ({
|
||||
companyId: context.fixture.companyId, endpointId: context.endpoint.id,
|
||||
conversationId: context.conversation.id, issueId: context.conversation.issueId,
|
||||
idempotencyKey: `run:${id}:working:${context.endpoint.id}`,
|
||||
payload: { text: "Maya is working…", progressState: "working" },
|
||||
state: "published", createdAt: valid.baseCreatedAt,
|
||||
})));
|
||||
const eventAt = new Date(valid.baseCreatedAt.getTime() + 30_000);
|
||||
await db.insert(heartbeatRunEvents).values(rejectedIds.map(runId => ({
|
||||
companyId: source.companyId, runId, agentId: source.agentId,
|
||||
seq: 1, eventType: "item.completed",
|
||||
createdAt: sql`${eventAt.toISOString()}::timestamptz + interval '0.000123 seconds'`,
|
||||
})));
|
||||
await context.addEvent(valid, "item.completed", 1, new Date(eventAt.getTime() + 1));
|
||||
const select = db.select.bind(db);
|
||||
let pages = 0;
|
||||
const guard = vi.spyOn(db, "select").mockImplementation((...args) => {
|
||||
if (args[0] && "eventCreatedAt" in args[0] && ++pages > 10) {
|
||||
throw new Error("Native progress pagination did not advance past suppressed candidates");
|
||||
}
|
||||
return select(...args);
|
||||
});
|
||||
try {
|
||||
await expect(enqueueSafeNativeChatProgress(db, { since: eventAt, limit: 1 })).resolves.toBe(1);
|
||||
} finally { guard.mockRestore(); }
|
||||
const progress = await db.select().from(chatPublications).where(and(
|
||||
eq(chatPublications.conversationId, context.conversation.id),
|
||||
like(chatPublications.idempotencyKey, "%:native:%"),
|
||||
));
|
||||
expect(progress).toEqual([expect.objectContaining({
|
||||
idempotencyKey: `run:${valid.runId}:working:${context.endpoint.id}:native:making_progress:1`,
|
||||
})]);
|
||||
} finally { await retirePublicationFixture(context.service, context.endpoint.id); }
|
||||
});
|
||||
|
||||
it.each(["separate owners", "combined budget"] as const)(
|
||||
"keeps lifecycle and native progress projection independent with %s",
|
||||
async (mode) => {
|
||||
|
||||
@@ -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([]);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -1,16 +1,18 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { eq, inArray, sql } from "drizzle-orm";
|
||||
import { drizzle } from "drizzle-orm/postgres-js";
|
||||
import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATOMIC_EVENTS_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
describeEmbeddedPostgres("atomic ordinary run events", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let otherDb: ReturnType<typeof createDb>;
|
||||
const companyId = "30000000-0000-4000-8000-000000000001";
|
||||
@@ -30,9 +32,10 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
|
||||
.where(eq(heartbeatRunEvents.runId, id)).orderBy(heartbeatRunEvents.seq);
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-");
|
||||
db = createDb(temporary.connectionString);
|
||||
otherDb = createDb(temporary.connectionString);
|
||||
if (!externalTestDatabaseUrl) temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-");
|
||||
const connectionString = externalTestDatabaseUrl ?? temporary!.connectionString;
|
||||
db = createDb(connectionString);
|
||||
otherDb = createDb(connectionString);
|
||||
await db.insert(companies).values([
|
||||
{ id: companyId, name: "Atomic events", issuePrefix: "ATM" },
|
||||
{ id: otherCompanyId, name: "Other events", issuePrefix: "OTH" },
|
||||
@@ -42,7 +45,21 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
|
||||
{ id: otherAgentId, companyId: otherCompanyId, name: "Other agent" },
|
||||
]);
|
||||
}, 60_000);
|
||||
afterAll(async () => { await temporary?.cleanup(); });
|
||||
afterAll(async () => {
|
||||
try {
|
||||
if (externalTestDatabaseUrl && db) {
|
||||
const fixtureCompanies = [companyId, otherCompanyId];
|
||||
await db.delete(heartbeatRunEvents).where(inArray(heartbeatRunEvents.companyId, fixtureCompanies));
|
||||
await db.delete(heartbeatRuns).where(inArray(heartbeatRuns.companyId, fixtureCompanies));
|
||||
await db.delete(agents).where(inArray(agents.companyId, fixtureCompanies));
|
||||
await db.delete(companies).where(inArray(companies.id, fixtureCompanies));
|
||||
}
|
||||
} finally {
|
||||
await otherDb?.$client.end({ timeout: 1 });
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
await temporary?.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it("persists a decoded receipt in one executed statement", async () => {
|
||||
const runId = await makeRun();
|
||||
|
||||
@@ -270,7 +270,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, publication.conversationId);
|
||||
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({ companyId: scope.companyId, issueId: scope.issueId }));
|
||||
}
|
||||
@@ -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 };
|
||||
}
|
||||
@@ -0,0 +1,427 @@
|
||||
import { and, asc, desc, eq, gt, gte, inArray, like, notExists, sql } from "drizzle-orm";
|
||||
import { alias } from "drizzle-orm/pg-core";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents, chatConversations, chatEndpoints, chatPublications, heartbeatRunEvents, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
|
||||
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, type ChatRunProjectionScope } from "./chat-run-publication-policy.js";
|
||||
import { SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, safeNativeChatProgressForEvent } from "./safe-native-chat-progress.js";
|
||||
|
||||
type SafeNativeChatProgressCandidate = {
|
||||
eventId: number;
|
||||
eventSeq: number;
|
||||
eventType: string;
|
||||
eventCreatedAt: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
companyId: string;
|
||||
endpointId: string;
|
||||
conversationId: string;
|
||||
agentName: string;
|
||||
};
|
||||
|
||||
const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000;
|
||||
|
||||
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",
|
||||
);
|
||||
const laterEvent = alias(
|
||||
heartbeatRunEvents,
|
||||
"safe_native_progress_later_event",
|
||||
);
|
||||
const issueIdFromContext = boundRunIssueIdSql();
|
||||
return async (pageSize: number, pageCursor: NativeProgressCursor | null): Promise<SafeNativeChatProgressCandidate[]> => await db
|
||||
.select({
|
||||
eventId: heartbeatRunEvents.id,
|
||||
eventSeq: heartbeatRunEvents.seq,
|
||||
eventType: heartbeatRunEvents.eventType,
|
||||
// Keep PostgreSQL microseconds in the cursor. Date truncation can
|
||||
// repeatedly select the same suppressed page without advancing.
|
||||
eventCreatedAt: sql<string>`${heartbeatRunEvents.createdAt}::text`,
|
||||
runId: heartbeatRuns.id,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
issueId: chatConversations.issueId,
|
||||
companyId: chatConversations.companyId,
|
||||
endpointId: chatConversations.endpointId,
|
||||
conversationId: chatConversations.id,
|
||||
agentName: agents.name,
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.innerJoin(
|
||||
heartbeatRuns,
|
||||
and(
|
||||
eq(heartbeatRuns.id, heartbeatRunEvents.runId),
|
||||
eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId),
|
||||
eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
chatConversations,
|
||||
and(
|
||||
eq(chatConversations.companyId, heartbeatRuns.companyId),
|
||||
sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`,
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
chatEndpoints,
|
||||
and(
|
||||
eq(chatEndpoints.companyId, chatConversations.companyId),
|
||||
eq(chatEndpoints.id, chatConversations.endpointId),
|
||||
eq(chatEndpoints.publicationMode, "automatic"),
|
||||
allowsAutomaticRunPublication(),
|
||||
eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId),
|
||||
),
|
||||
)
|
||||
.innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
|
||||
.innerJoin(
|
||||
basePublication,
|
||||
and(
|
||||
eq(basePublication.companyId, heartbeatRuns.companyId),
|
||||
eq(basePublication.endpointId, chatConversations.endpointId),
|
||||
eq(basePublication.conversationId, chatConversations.id),
|
||||
eq(basePublication.issueId, chatConversations.issueId),
|
||||
sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`,
|
||||
inArray(basePublication.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"published",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
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
|
||||
.select({ id: laterEvent.id })
|
||||
.from(laterEvent)
|
||||
.where(
|
||||
and(
|
||||
eq(laterEvent.companyId, heartbeatRunEvents.companyId),
|
||||
eq(laterEvent.runId, heartbeatRunEvents.runId),
|
||||
gt(laterEvent.seq, heartbeatRunEvents.seq),
|
||||
inArray(laterEvent.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
),
|
||||
),
|
||||
),
|
||||
pageCursor
|
||||
? sql`(${heartbeatRunEvents.createdAt}, ${heartbeatRuns.id}, ${chatConversations.id}) >
|
||||
(${pageCursor.eventCreatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)`
|
||||
: undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(
|
||||
asc(heartbeatRunEvents.createdAt),
|
||||
asc(heartbeatRuns.id),
|
||||
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 = {
|
||||
eventCreatedAt: lastRow.eventCreatedAt,
|
||||
runId: lastRow.runId,
|
||||
conversationId: lastRow.conversationId,
|
||||
};
|
||||
|
||||
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
|
||||
// publication insert takes an issue FK lock, so locking the run first
|
||||
// would invert that order against a concurrent terminal commit. A
|
||||
// contended issue/run is optional progress, not a reason to hold other
|
||||
// chats: skip it now and revisit it with a fresh cursor next sweep.
|
||||
const currentIssue = await tx
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, row.issueId),
|
||||
eq(issues.companyId, row.companyId),
|
||||
),
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
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.
|
||||
const currentRun = 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"),
|
||||
),
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
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 })
|
||||
.from(chatConversations)
|
||||
.innerJoin(
|
||||
chatEndpoints,
|
||||
and(
|
||||
eq(chatEndpoints.companyId, chatConversations.companyId),
|
||||
eq(chatEndpoints.id, chatConversations.endpointId),
|
||||
eq(chatEndpoints.publicationMode, "automatic"),
|
||||
allowsAutomaticRunPublication(),
|
||||
eq(chatEndpoints.assignedAgentId, row.agentId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, row.companyId),
|
||||
eq(chatConversations.id, row.conversationId),
|
||||
eq(chatConversations.endpointId, row.endpointId),
|
||||
eq(chatConversations.issueId, row.issueId),
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!destination) return 0;
|
||||
|
||||
const [currentEvent] = await tx
|
||||
.select({
|
||||
id: heartbeatRunEvents.id,
|
||||
seq: heartbeatRunEvents.seq,
|
||||
eventType: heartbeatRunEvents.eventType,
|
||||
createdAt: heartbeatRunEvents.createdAt,
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRunEvents.companyId, row.companyId),
|
||||
eq(heartbeatRunEvents.runId, row.runId),
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id))
|
||||
.limit(1);
|
||||
if (
|
||||
!currentEvent ||
|
||||
currentEvent.id !== row.eventId ||
|
||||
currentEvent.seq !== row.eventSeq ||
|
||||
currentEvent.eventType !== row.eventType
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const progress = safeNativeChatProgressForEvent(
|
||||
currentEvent.eventType,
|
||||
row.agentName,
|
||||
);
|
||||
if (!progress) return 0;
|
||||
|
||||
const baseKey = `run:${row.runId}:working:${row.endpointId}`;
|
||||
const [currentBase] = await tx
|
||||
.select({ createdAt: chatPublications.createdAt })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
eq(chatPublications.idempotencyKey, baseKey),
|
||||
inArray(chatPublications.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"published",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (!currentBase) return 0;
|
||||
|
||||
const progressKeyPrefix = `${baseKey}:native:`;
|
||||
const previousProgress = await tx
|
||||
.select({
|
||||
createdAt: chatPublications.createdAt,
|
||||
idempotencyKey: chatPublications.idempotencyKey,
|
||||
})
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`),
|
||||
),
|
||||
);
|
||||
if (
|
||||
previousProgress.some((publication) =>
|
||||
publication.idempotencyKey.startsWith(
|
||||
`${progressKeyPrefix}${progress.phase}:`,
|
||||
),
|
||||
)
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const lastProgressAt = Math.max(
|
||||
currentBase.createdAt.getTime(),
|
||||
...previousProgress.map((publication) =>
|
||||
publication.createdAt.getTime(),
|
||||
),
|
||||
);
|
||||
if (
|
||||
currentEvent.createdAt.getTime() - lastProgressAt <
|
||||
SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const bindings = await resolveChatOriginPublicationBindings(
|
||||
tx,
|
||||
row.companyId,
|
||||
row.issueId,
|
||||
row.runId,
|
||||
);
|
||||
if (
|
||||
!bindings.some(
|
||||
(binding) =>
|
||||
binding.endpointId === row.endpointId &&
|
||||
binding.conversationId === row.conversationId,
|
||||
)
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
if (
|
||||
await hasChatRunOwnedProviderInteraction(tx, {
|
||||
companyId: row.companyId,
|
||||
issueId: row.issueId,
|
||||
runId: row.runId,
|
||||
})
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const explicitlyAuthoredFinal = await tx
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.innerJoin(
|
||||
issueComments,
|
||||
and(
|
||||
eq(issueComments.companyId, chatPublications.companyId),
|
||||
eq(issueComments.id, chatPublications.commentId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
eq(issueComments.authorType, "agent"),
|
||||
eq(issueComments.createdByRunId, row.runId),
|
||||
sql`(
|
||||
${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
|
||||
or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
|
||||
)`,
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (explicitlyAuthoredFinal.length > 0) return 0;
|
||||
|
||||
await notifyChatPublicationWork(tx);
|
||||
const insertedRows = await tx
|
||||
.insert(chatPublications)
|
||||
.values({
|
||||
companyId: row.companyId,
|
||||
endpointId: row.endpointId,
|
||||
conversationId: row.conversationId,
|
||||
issueId: row.issueId,
|
||||
idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`,
|
||||
payload: projectSafeChatPublication({
|
||||
classification: "external",
|
||||
source: "safe_milestone",
|
||||
text: progress.text,
|
||||
progressState: "working",
|
||||
}),
|
||||
state: "pending",
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatPublications.id });
|
||||
return insertedRows.length;
|
||||
});
|
||||
inserted += result;
|
||||
}
|
||||
if (rows.length < pageSize) break;
|
||||
}
|
||||
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);
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import { ne, notExists, or, sql } from "drizzle-orm";
|
||||
import { chatEndpoints, chatGitHubConfigurations } from "@paperclipai/db";
|
||||
|
||||
// Only upgraded GitHub bots have the task-scoped response tools. Legacy
|
||||
// connections retain automatic progress and final-answer publication.
|
||||
export function allowsAutomaticRunPublication() {
|
||||
return or(ne(chatEndpoints.provider, "github"), notExists(
|
||||
sql`(select 1 from ${chatGitHubConfigurations}
|
||||
where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId}
|
||||
and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`,
|
||||
));
|
||||
}
|
||||
|
||||
export function chatRunProjectionWindow(input: { since?: Date; limit?: number }) {
|
||||
return {
|
||||
since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000),
|
||||
limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)),
|
||||
};
|
||||
}
|
||||
|
||||
export type ChatRunProjectionScope = { companyId: string; issueId?: string };
|
||||
@@ -1,21 +1,22 @@
|
||||
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";
|
||||
import type { ChatRunMilestoneScope } from "./chat-run-milestone-notifications.js";
|
||||
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
desc,
|
||||
eq,
|
||||
gt,
|
||||
gte,
|
||||
inArray,
|
||||
isNull,
|
||||
like,
|
||||
ne,
|
||||
notExists,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
import { alias } from "drizzle-orm/pg-core";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
@@ -26,7 +27,6 @@ import {
|
||||
chatGitHubReviews,
|
||||
chatMessageLinks,
|
||||
chatPublications,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
issues,
|
||||
@@ -39,10 +39,6 @@ import { resolveChatOriginPublicationBindings } from "./issues.js";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { chatRunMilestoneSource } from "./chat-run-milestone-candidates.js";
|
||||
import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js";
|
||||
import {
|
||||
SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
safeNativeChatProgressForEvent,
|
||||
} from "./safe-native-chat-progress.js";
|
||||
|
||||
type SafeRunMilestone =
|
||||
"queued" | "working" | "waiting_for_input" | "completed" | "failed";
|
||||
@@ -54,15 +50,6 @@ const OWNERSHIP_ATTENTION_CODES = [
|
||||
|
||||
export { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON };
|
||||
|
||||
// Only upgraded GitHub bots have the task-scoped response tools. Legacy
|
||||
// connections retain automatic progress and final-answer publication.
|
||||
function allowsAutomaticRunPublication() {
|
||||
return or(ne(chatEndpoints.provider, "github"), notExists(
|
||||
sql`(select 1 from ${chatGitHubConfigurations}
|
||||
where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId}
|
||||
and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`,
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* Heartbeat's presentation resolver may externalize its selected final prose
|
||||
@@ -227,21 +214,6 @@ type ChatRunMilestoneCandidate = {
|
||||
agentName: string;
|
||||
};
|
||||
|
||||
type SafeNativeChatProgressCandidate = {
|
||||
eventId: number;
|
||||
eventSeq: number;
|
||||
eventType: string;
|
||||
eventCreatedAt: Date;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
companyId: string;
|
||||
endpointId: string;
|
||||
conversationId: string;
|
||||
agentName: string;
|
||||
};
|
||||
|
||||
const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000;
|
||||
|
||||
function milestoneForStatus(
|
||||
status: string,
|
||||
@@ -293,384 +265,6 @@ export function safeMilestoneText(input: {
|
||||
}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Projects a bounded sample of native activity into the existing run working
|
||||
* lane. The selector intentionally reads only event identity, type, sequence,
|
||||
* and time; native messages and payloads stay inside Paperclip.
|
||||
*/
|
||||
export async function enqueueSafeNativeChatProgress(
|
||||
db: Db,
|
||||
options: { since?: Date; limit?: number } = {},
|
||||
): Promise<number> {
|
||||
const input = chatRunProjectionWindow(options);
|
||||
const basePublication = alias(
|
||||
chatPublications,
|
||||
"safe_native_progress_base_publication",
|
||||
);
|
||||
const laterEvent = alias(
|
||||
heartbeatRunEvents,
|
||||
"safe_native_progress_later_event",
|
||||
);
|
||||
const issueIdFromContext = boundRunIssueIdSql();
|
||||
let inserted = 0;
|
||||
let cursor: {
|
||||
eventCreatedAt: Date;
|
||||
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
|
||||
.select({
|
||||
eventId: heartbeatRunEvents.id,
|
||||
eventSeq: heartbeatRunEvents.seq,
|
||||
eventType: heartbeatRunEvents.eventType,
|
||||
eventCreatedAt: heartbeatRunEvents.createdAt,
|
||||
runId: heartbeatRuns.id,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
issueId: chatConversations.issueId,
|
||||
companyId: chatConversations.companyId,
|
||||
endpointId: chatConversations.endpointId,
|
||||
conversationId: chatConversations.id,
|
||||
agentName: agents.name,
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.innerJoin(
|
||||
heartbeatRuns,
|
||||
and(
|
||||
eq(heartbeatRuns.id, heartbeatRunEvents.runId),
|
||||
eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId),
|
||||
eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
chatConversations,
|
||||
and(
|
||||
eq(chatConversations.companyId, heartbeatRuns.companyId),
|
||||
sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`,
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
chatEndpoints,
|
||||
and(
|
||||
eq(chatEndpoints.companyId, chatConversations.companyId),
|
||||
eq(chatEndpoints.id, chatConversations.endpointId),
|
||||
eq(chatEndpoints.publicationMode, "automatic"),
|
||||
allowsAutomaticRunPublication(),
|
||||
eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId),
|
||||
),
|
||||
)
|
||||
.innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
|
||||
.innerJoin(
|
||||
basePublication,
|
||||
and(
|
||||
eq(basePublication.companyId, heartbeatRuns.companyId),
|
||||
eq(basePublication.endpointId, chatConversations.endpointId),
|
||||
eq(basePublication.conversationId, chatConversations.id),
|
||||
eq(basePublication.issueId, chatConversations.issueId),
|
||||
sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`,
|
||||
inArray(basePublication.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"published",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
gte(heartbeatRunEvents.createdAt, input.since),
|
||||
sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`,
|
||||
notExists(
|
||||
db
|
||||
.select({ id: laterEvent.id })
|
||||
.from(laterEvent)
|
||||
.where(
|
||||
and(
|
||||
eq(laterEvent.companyId, heartbeatRunEvents.companyId),
|
||||
eq(laterEvent.runId, heartbeatRunEvents.runId),
|
||||
gt(laterEvent.seq, heartbeatRunEvents.seq),
|
||||
inArray(laterEvent.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
),
|
||||
),
|
||||
),
|
||||
pageCursor
|
||||
? or(
|
||||
gt(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt),
|
||||
and(
|
||||
eq(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt),
|
||||
or(
|
||||
gt(heartbeatRuns.id, pageCursor.runId),
|
||||
and(
|
||||
eq(heartbeatRuns.id, pageCursor.runId),
|
||||
gt(chatConversations.id, pageCursor.conversationId),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
: undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(
|
||||
asc(heartbeatRunEvents.createdAt),
|
||||
asc(heartbeatRuns.id),
|
||||
asc(chatConversations.id),
|
||||
)
|
||||
.limit(pageSize);
|
||||
if (rows.length === 0) break;
|
||||
const lastRow = rows.at(-1)!;
|
||||
cursor = {
|
||||
eventCreatedAt: lastRow.eventCreatedAt,
|
||||
runId: lastRow.runId,
|
||||
conversationId: lastRow.conversationId,
|
||||
};
|
||||
|
||||
for (const row of rows) {
|
||||
if (inserted >= input.limit) break;
|
||||
const result = await db.transaction(async (tx) => {
|
||||
// Keep the normal issue -> run order used by native finalization. The
|
||||
// publication insert takes an issue FK lock, so locking the run first
|
||||
// would invert that order against a concurrent terminal commit. A
|
||||
// contended issue/run is optional progress, not a reason to hold other
|
||||
// chats: skip it now and revisit it with a fresh cursor next sweep.
|
||||
const currentIssue = await tx
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, row.issueId),
|
||||
eq(issues.companyId, row.companyId),
|
||||
),
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!currentIssue) 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.
|
||||
const currentRun = 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"),
|
||||
),
|
||||
)
|
||||
.for("update", { skipLocked: true })
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!currentRun) return 0;
|
||||
|
||||
const destination = await tx
|
||||
.select({ id: chatConversations.id })
|
||||
.from(chatConversations)
|
||||
.innerJoin(
|
||||
chatEndpoints,
|
||||
and(
|
||||
eq(chatEndpoints.companyId, chatConversations.companyId),
|
||||
eq(chatEndpoints.id, chatConversations.endpointId),
|
||||
eq(chatEndpoints.publicationMode, "automatic"),
|
||||
allowsAutomaticRunPublication(),
|
||||
eq(chatEndpoints.assignedAgentId, row.agentId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, row.companyId),
|
||||
eq(chatConversations.id, row.conversationId),
|
||||
eq(chatConversations.endpointId, row.endpointId),
|
||||
eq(chatConversations.issueId, row.issueId),
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((currentRows) => currentRows[0] ?? null);
|
||||
if (!destination) return 0;
|
||||
|
||||
const [currentEvent] = await tx
|
||||
.select({
|
||||
id: heartbeatRunEvents.id,
|
||||
seq: heartbeatRunEvents.seq,
|
||||
eventType: heartbeatRunEvents.eventType,
|
||||
createdAt: heartbeatRunEvents.createdAt,
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRunEvents.companyId, row.companyId),
|
||||
eq(heartbeatRunEvents.runId, row.runId),
|
||||
inArray(heartbeatRunEvents.eventType, [
|
||||
...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id))
|
||||
.limit(1);
|
||||
if (
|
||||
!currentEvent ||
|
||||
currentEvent.id !== row.eventId ||
|
||||
currentEvent.seq !== row.eventSeq ||
|
||||
currentEvent.eventType !== row.eventType
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const progress = safeNativeChatProgressForEvent(
|
||||
currentEvent.eventType,
|
||||
row.agentName,
|
||||
);
|
||||
if (!progress) return 0;
|
||||
|
||||
const baseKey = `run:${row.runId}:working:${row.endpointId}`;
|
||||
const [currentBase] = await tx
|
||||
.select({ createdAt: chatPublications.createdAt })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
eq(chatPublications.idempotencyKey, baseKey),
|
||||
inArray(chatPublications.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"published",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (!currentBase) return 0;
|
||||
|
||||
const progressKeyPrefix = `${baseKey}:native:`;
|
||||
const previousProgress = await tx
|
||||
.select({
|
||||
createdAt: chatPublications.createdAt,
|
||||
idempotencyKey: chatPublications.idempotencyKey,
|
||||
})
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`),
|
||||
),
|
||||
);
|
||||
if (
|
||||
previousProgress.some((publication) =>
|
||||
publication.idempotencyKey.startsWith(
|
||||
`${progressKeyPrefix}${progress.phase}:`,
|
||||
),
|
||||
)
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const lastProgressAt = Math.max(
|
||||
currentBase.createdAt.getTime(),
|
||||
...previousProgress.map((publication) =>
|
||||
publication.createdAt.getTime(),
|
||||
),
|
||||
);
|
||||
if (
|
||||
currentEvent.createdAt.getTime() - lastProgressAt <
|
||||
SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const bindings = await resolveChatOriginPublicationBindings(
|
||||
tx,
|
||||
row.companyId,
|
||||
row.issueId,
|
||||
row.runId,
|
||||
);
|
||||
if (
|
||||
!bindings.some(
|
||||
(binding) =>
|
||||
binding.endpointId === row.endpointId &&
|
||||
binding.conversationId === row.conversationId,
|
||||
)
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
if (
|
||||
await hasChatRunOwnedProviderInteraction(tx, {
|
||||
companyId: row.companyId,
|
||||
issueId: row.issueId,
|
||||
runId: row.runId,
|
||||
})
|
||||
) {
|
||||
return 0;
|
||||
}
|
||||
const explicitlyAuthoredFinal = await tx
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.innerJoin(
|
||||
issueComments,
|
||||
and(
|
||||
eq(issueComments.companyId, chatPublications.companyId),
|
||||
eq(issueComments.id, chatPublications.commentId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, row.companyId),
|
||||
eq(chatPublications.endpointId, row.endpointId),
|
||||
eq(chatPublications.conversationId, row.conversationId),
|
||||
eq(chatPublications.issueId, row.issueId),
|
||||
eq(issueComments.authorType, "agent"),
|
||||
eq(issueComments.createdByRunId, row.runId),
|
||||
sql`(
|
||||
${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
|
||||
or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
|
||||
)`,
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (explicitlyAuthoredFinal.length > 0) return 0;
|
||||
|
||||
await notifyChatPublicationWork(tx);
|
||||
const insertedRows = await tx
|
||||
.insert(chatPublications)
|
||||
.values({
|
||||
companyId: row.companyId,
|
||||
endpointId: row.endpointId,
|
||||
conversationId: row.conversationId,
|
||||
issueId: row.issueId,
|
||||
idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`,
|
||||
payload: projectSafeChatPublication({
|
||||
classification: "external",
|
||||
source: "safe_milestone",
|
||||
text: progress.text,
|
||||
progressState: "working",
|
||||
}),
|
||||
state: "pending",
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatPublications.id });
|
||||
return insertedRows.length;
|
||||
});
|
||||
inserted += result;
|
||||
}
|
||||
if (rows.length < pageSize) break;
|
||||
}
|
||||
return inserted;
|
||||
}
|
||||
|
||||
type ChatRunMilestoneCursor = {
|
||||
updatedAt: string;
|
||||
@@ -1096,6 +690,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)
|
||||
@@ -1135,12 +732,6 @@ type ChatRunProjectionInput = {
|
||||
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. */
|
||||
|
||||
@@ -57,6 +57,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,10 @@ 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,
|
||||
// Only progress needs the task binding. Other native/recovery events
|
||||
// must not read a potentially large context snapshot to append a log.
|
||||
...(projectsProgress ? { runtimeMode: heartbeatRuns.runtimeMode, issueId: boundRunIssueIdSql() } : {}) })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.for("update")
|
||||
@@ -165,6 +172,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