Repository navigation
feat(core): sync engine and paused workflow runs (delay, approval) for integration definitions - #695
Conversation
Data-defined syncs now run: backfill, webhook and polling updates, both directions for two-way syncs, per-field master rules, echo suppression, and relinking after an identity change. - Manifests declare how a sync reads and writes each object (objects.<obj>.sync: list with cursor/offset/page paging, get, search, create, update, body wrapper, id and updatedAt paths), plus recordId on events that carry only an id. validateManifest and the JSON Schema check them; sync blocks added for the 16 objects the examples sync. - Definitions refuse syncs whose sides cannot do what is asked, and may pass per-module arguments (sync.args, $.install only). New template filters: split, last. - sync-engine.js: link -> identity index -> search (normalised keys, onAmbiguous) -> create/update; master source/target/newest on update; writes only changed fields that the capability accepts; hashed known values per side end the echo; backfill pages with a progress row, one run at a time, failures counted; polls with a high-water mark and a page cap. - The class runs SYNC_*_BACKFILL/PAGE/ONGOING through the queue, starts onInstall backfills, relinks on upgrade, fires syncRecord workflows (created/linked once per record), and serves GET /api/integrations/:id/syncs. 156 data-defined tests pass on real PostgreSQL and MongoDB; 23 mutations killed (one dead branch removed instead). Core's full suite fails the same 16 suites / 65 tests as before (pre-existing). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
Two built-in steps end a run for now with status `waiting` and carry it
on later from the next step.
- delay { for | until } and approval { prompt, approvers, timeout,
onTimeout } (top level of a workflow only; validation and the
interpreter both refuse them elsewhere).
- The checkpoint (step index, step results, vars, event, trace so far)
is an encrypted IntegrationMapping row; the WorkflowRun row gets
runKey, waitingFor and resumeAt (new columns, migration
20261019000000_workflow_run_waits) and stays one row across pauses.
A checkpoint that cannot be saved gives the event's dedupe claim back.
- Resume: the delay's end or the approval's timeout (one-time scheduler
job), or a decision via POST /api/integrations/:id/runs/:runId/decision
(202, through the queue). One claim per pause, so a race resumes once;
stale or wrong-kind resumes are ignored; a failed resume gives the claim
back for the retry; the run finishes on the version it started on.
- WorkflowRun repository: findRun, updateRun (Mongo, Postgres,
DocumentDB).
173 data-defined tests pass on real PostgreSQL and MongoDB; 23
mutations killed. Core's full suite fails the same 16 suites / 65 tests
as before (pre-existing).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
If your organization's extra usage balance is empty, an organization admin can add extra usage credits at claude.ai/admin-settings/usage. If its monthly spend limit was reached, an admin can raise it on the same page. If neither applies, contact Anthropic support.
Once extra usage is available, someone with write access to this repository can comment @claude review on this pull request to trigger a review.
seanspeaks
added
release
Claude
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| await this.fireSyncWorkflows(ref.id, [result]); | ||
| return result; | ||
| } | ||
| return engine.poll({ syncId: ref.id }); |
There was a problem hiding this comment.
Poll skips syncRecord workflows
High Severity
A poll tick syncs records through readPage / syncRecord but never calls fireSyncWorkflows. Only webhook ongoing events enqueue created / updated / linked runs, so poll-mode syncs silently drop those workflows.
Reviewed by Cursor Bugbot for commit efa0b08. Configure here.
| const run = await resumeWorkflow({ def, installation: this.installation(), checkpoint: cp, result, stop, ports: def === this.def ? this.ports : this.portsFor(def), runId }); | ||
| if (run.status === 'failed') await store.release(claimKey); // the queue's retry may resume again | ||
| else if (run.status !== 'waiting') await store.release(`wait:${runId}`); | ||
| return this.recordRun(run, { rowId: cp.rowId }); |
There was a problem hiding this comment.
Resume claim leaked on throw
High Severity
resumeRun takes the resume:<runId>:<index> claim before loading the definition or calling resumeWorkflow, and only releases it when the run object comes back as failed. A throw from definitions.get or resumeWorkflow leaves the claim in place, so queue retries return duplicate and the pause never continues.
Reviewed by Cursor Bugbot for commit efa0b08. Configure here.
| known: { ...(link?.known || {}), [A.key]: knownA, [B.key]: { ...knownB, ...written } }, | ||
| seen: { ...(link?.seen || {}), ...(updatedAt ? { [A.key]: updatedAt } : {}), ...(wrote ? { [B.key]: now().toISOString() } : {}) }, | ||
| }, | ||
| }); |
There was a problem hiding this comment.
Stale identity keys stay indexed
Medium Severity
When a synced record’s identity values change, links.upsert writes the new idx: rows and leaves the old ones in place. A later record that presents the retired email or phone can match the previous pair and overwrite the wrong remote record.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit efa0b08. Configure here.
|
ADR reconciliation needed after Stack A/B — drafted with Claude Code at Sean's request Severity: must-reconcile. This PR adds a second sync engine and a small workflow engine next to ADR-044/047 (#648) and ADR-054's experimental sync pieces (#669). The ADR text changes are proposed on #648 and #669; these are the items for this PR. Overlaps
Proposed
|
…ator - startBackfill: build the run id suffix from crypto.randomBytes instead of Math.random (S2245); same bf_<time>_<6-8 char> shape, now 8 hex chars. - fireSyncWorkflows: sort dedupe id parts with an explicit code-unit comparator (S2871); order is unchanged so existing dedupe ids still match. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
- 15b: this engine is the first Tier 2-3 implementation of ADR-044, the canonical spec, and moves to packages/core/syncs/ later; the Lambda recursion override is interim until ADR-047's orchestrator queue. - 15c: WorkflowRun is an ADR-047 Tier-3 instance row; deciding an approval should require manage access under ADR-056 (follow-up). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
…s; stack-4 sync poll adapted to triggers, stack-4 test names neutralised) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NGC9SnFuqdjX2xH2qeu8vw
|
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 4 potential issues.
There are 7 total unresolved issues (including 3 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 5edaa52. Configure here.
| const integration = await getIntegrationInstance.execute(req.params.integrationId, user.getId()); | ||
| if (!integration?.constructor?.DataDefined) throw new DefinitionError('This integration is not data-defined', { status: 404, code: 'NOT_DATA_DEFINED' }); | ||
| const out = await integration.decide(req.params.runId, { decision: req.body?.decision, note: req.body?.note, userId: user.getId() }); | ||
| res.status(202).json(out); |
There was a problem hiding this comment.
Decision route uses wrong run id
High Severity
GET /runs?status=waiting returns database id as the run identifier, while POST /runs/:runId/decision looks up the checkpoint as wait:<runId> using the runtime runKey. Listing pending approvals and posting a decision with the listed id yields 404 even though the run is waiting.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit 5edaa52. Configure here.
| seen: { ...(link?.seen || {}), ...(updatedAt ? { [A.key]: updatedAt } : {}), ...(wrote ? { [B.key]: now().toISOString() } : {}) }, | ||
| }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
Relink leaves identity index stale
Medium Severity
A relink backfill does not rewrite identity indexes for records that already link and have not changed. syncRecord never sees the relink reason and skips links.upsert when changedKnowledge is false, so new identity keys from an upgrade are never written and lookups by those keys miss.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 5edaa52. Configure here.
| let next = null; | ||
| if (mode === 'cursor') { const n = L.next ? readField(res, L.next) : null; next = blank(n) ? null : n; } | ||
| else if (list.length >= pageSize) next = mode === 'offset' ? (Number(cursor) || 0) + list.length : (Number(cursor) || 1) + 1; | ||
| if (next !== null && String(next) === String(cursor)) next = null; // a cursor that does not move would loop |
There was a problem hiding this comment.
QuickBooks page offset is off-by-one
Medium Severity
Offset paging treats a missing cursor as 0 and advances by page length, but QuickBooks startPosition is 1-based and defaults to 1. The next page is requested at the last index of the previous page, so every boundary record is synced twice and later windows shift.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 5edaa52. Configure here.
| if (following) Object.assign(updated, { role: following.role, cursor: following.cursor }); | ||
| else Object.assign(updated, { status: 'completed', finishedAt: updated.updatedAt }); | ||
| await state.put(backfillKey(s.id), updated); | ||
| return { progress: updated, next: following, results }; |
There was a problem hiding this comment.
Backfill pages replay under redelivery
Medium Severity
continueBackfill only checks that the progress runId is still running. It does not claim the page or require the message cursor to match progress.cursor. A redelivered PAGE message for an earlier cursor re-reads that page while the backfill is still running, which can create duplicate target records when two workers race before the link is written.
Reviewed by Cursor Bugbot for commit 5edaa52. Configure here.
|
Release review follow-up (folded in at Done
Waiting
Generated by Claude Code |
Stack B review (COMMENT only) — Group 4Verdict (would merge?): changes needed Blocking
Should-fix
TestsSync/pause suites are solid; ownership gap is what the fake |
| pages += 1; | ||
| } while (cursor !== null && pages < maxPages); | ||
| if (cursor !== null) st.cursor[role] = cursor; // more to read: same mark, carry on next tick | ||
| else { delete st.cursor[role]; st.since[role] = newest ? new Date(Date.parse(newest)).toISOString() : started; } |
There was a problem hiding this comment.
[should-fix] Poll advances since to newest even when the page had failed sync results.
Failure scenario: Poll page syncs 100 records; 3 fail; watermark moves past them → those 3 are skipped until they change again. Advance since only when the page had no failures, or keep a retry watermark.
|
|
||
| router.route('/api/integrations/:integrationId/runs/:runId/decision').post(guard(async (req, res) => { | ||
| const user = await authenticateUser.execute(req); | ||
| const integration = await getIntegrationInstance.execute(req.params.integrationId, user.getId()); |
There was a problem hiding this comment.
[blocking] Decision HTTP path loads the installation as the caller (getIntegrationInstance.execute(id, user.getId())), so only the owner can reach decide().
Failure scenario: Approval step sets approvers: ['boss']; installation owned by u1. boss POSTs /api/integrations/:id/runs/:runId/decision → ownership load fails (404). u1 reaches decide but gets 403 NOT_AN_APPROVER. Nobody can approve until timeout. Tests that stub getIntegrationInstance to always return the instance miss this. Load by id for decision (after auth), then enforce approvers (defaulting to owner when unset). (#696 introduces GetIntegrationForDecision — cherry-pick or land that before relying on named approvers here.)
| payload: { integrationId: this.id, resume: run.id, reason: cp.wait.kind === 'delay' ? 'delay' : 'timeout', index: cp.index }, | ||
| queueUrl: typeof this.scheduleQueueUrl === 'function' ? this.scheduleQueueUrl() : process.env[`${name.toUpperCase().replace(/-/g, '_')}_QUEUE_URL`], | ||
| }); | ||
| if (booked?.error || booked?.warning) { |
There was a problem hiding this comment.
[should-fix] When scheduleJob returns error/warning (scheduler disabled / IAM), the run stays waiting and only a log is written.
Failure scenario: Delay pause after a successful checkpoint write; booking fails → delay never resumes; approval timeouts also never fire. Prefer failing the pause (release dedupe / throw so the queue retries) or retry booking so delay/timeout runs are not stranded.
Stack 4 of 9. Stacked on #694.
Summary
Sync engine.
objects.<obj>.sync).onAmbiguousdecides what happens with several matches. Only fields that changed and that the target accepts are written.GET /api/integrations/:id/syncs.Paused runs. Two new steps,
delay { for | until }andapproval { prompt, approvers, timeout, onTimeout }, end a run with statuswaiting.IntegrationMappingrow.WorkflowRungainsrunKey,waitingForandresumeAt, with migration20261019000000_workflow_run_waits.POST /api/integrations/:id/runs/:runId/decision.Test plan
nextat this commit. Nothing new fails.Generated by Claude Code
Note
High Risk
Introduces bidirectional record sync, durable pause/resume state, and new authenticated decision routes—core integration behavior with data integrity and concurrency implications.
Overview
Data-defined integrations now run real syncs and can pause workflows mid-run, replacing the v0.1 stubs that threw on
runSync.A new
sync-engine.jsdrives definitionsyncs: paged backfills (queued one page per message), webhook and poll ongoing updates, two-way copy with per-field master rules and hash-based echo suppression. I/O is declared per object in capability manifests viaobjects.<obj>.sync(list/get/search/create/update); definitions can add per-modulesync.args. The integration class wiresSYNC_*handlers,onInstallbackfills, upgraderelink,syncRecordworkflow triggers, andGET /api/integrations/:id/syncsprogress.delayandapprovalare top-level workflow steps that end a run inwaiting: checkpoints live in encryptedwait:<runKey>mapping rows;WorkflowRungainsrunKey,waitingFor,resumeAt(Prisma migration). Resumes come from the scheduler orPOST …/runs/:runId/decision(202 + queue), with single-claim semantics and finish-on-pinned-definition-version behavior.Validation, manifest checks, mock capability JSON, ADR-063 docs, and a large test suite (sync engine, class wiring, pause/approval HTTP) accompany the change. v0.1 “not done” lists drop sync/delay/relink in favor of per-module rate limits still deferred.
Reviewed by Cursor Bugbot for commit 5edaa52. Bugbot is set up for automated code reviews on this repo. Configure here.
📦 Published PR as canary version:
2.0.0--canary.695.5edaa52.0✨ Test out this PR locally via:
npm install @friggframework/admin-scripts@2.0.0--canary.695.5edaa52.0 npm install @friggframework/core@2.0.0--canary.695.5edaa52.0 npm install @friggframework/devtools@2.0.0--canary.695.5edaa52.0 npm install @friggframework/eslint-config@2.0.0--canary.695.5edaa52.0 npm install @friggframework/prettier-config@2.0.0--canary.695.5edaa52.0 npm install @friggframework/schemas@2.0.0--canary.695.5edaa52.0 npm install @friggframework/serverless-plugin@2.0.0--canary.695.5edaa52.0 npm install @friggframework/test@2.0.0--canary.695.5edaa52.0 npm install @friggframework/ui@2.0.0--canary.695.5edaa52.0 # or yarn add @friggframework/admin-scripts@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/core@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/devtools@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/eslint-config@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/prettier-config@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/schemas@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/serverless-plugin@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/test@2.0.0--canary.695.5edaa52.0 yarn add @friggframework/ui@2.0.0--canary.695.5edaa52.0