diff --git a/.github/workflows/continuous-delivery-cloud.yml b/.github/workflows/continuous-delivery-cloud.yml index 2a7a8cc94246..9f6e1b1bb87e 100644 --- a/.github/workflows/continuous-delivery-cloud.yml +++ b/.github/workflows/continuous-delivery-cloud.yml @@ -160,6 +160,17 @@ jobs: git push origin "$BRANCH" --force echo "Created $BRANCH at $SHA" + - name: Notify on-call channel + env: + WEBHOOK: ${{ secrets.DISCORD_ON_CALL_WEBHOOK }} + IMAGE_TAG: ${{ steps.image.outputs.image_tag }} + DEPLOY_KIND: ${{ github.event_name == 'workflow_dispatch' && 'hotfix' || 'scheduled promotion' }} + run: | + curl -H "Content-Type: application/json" \ + -X POST \ + -d "{\"content\": \"πŸš€ **Cloud production deployed** ($DEPLOY_KIND)\\n\\n**Version:** \`$IMAGE_TAG\`\\n**Changelog:** ${{ github.server_url }}/${{ github.repository }}/releases/latest\\n**Run:** ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}\"}" \ + "$WEBHOOK" + # Run smoke test after scheduled promotion smoke-test: needs: promote-to-production diff --git a/.github/workflows/release-self-hosted.yml b/.github/workflows/release-self-hosted.yml index 6c2eb4cd4c55..ea83090d9d5b 100644 --- a/.github/workflows/release-self-hosted.yml +++ b/.github/workflows/release-self-hosted.yml @@ -115,6 +115,17 @@ jobs: image: ghcr.io/activepieces/activepieces:${{ inputs.tag }} version: ${{ inputs.tag }} + - name: Notify on-call channel + if: ${{ !contains(inputs.tag, '-rc') || inputs.publish_rc_release }} + env: + WEBHOOK: ${{ secrets.DISCORD_ON_CALL_WEBHOOK }} + TAG: ${{ inputs.tag }} + run: | + curl -H "Content-Type: application/json" \ + -X POST \ + -d "{\"content\": \"πŸ“¦ **Self-hosted $TAG released**\\n\\n**Docker image:** \`activepieces/activepieces:$TAG\` (also on ghcr.io)\\n**Changelog:** ${{ github.server_url }}/${{ github.repository }}/releases/tag/$TAG\\n**Run:** ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}\"}" \ + "$WEBHOOK" + sync-version-to-main: needs: release if: ${{ !cancelled() && needs.release.result == 'success' }} diff --git a/brain/knowledge/ai-intelligence/ai-agents.md b/brain/knowledge/ai-intelligence/ai-agents.md index 740fb93b1237..a0b0b7d0788d 100644 --- a/brain/knowledge/ai-intelligence/ai-agents.md +++ b/brain/knowledge/ai-intelligence/ai-agents.md @@ -28,6 +28,9 @@ A flow step type (backed by `@activepieces/piece-agent`) that runs an LLM-driven - **The enums and pure functions have exactly one home: `core/piece-types/src/lib/agents.ts`.** Do not re-declare `AgentToolType`, `McpAuthType`, `buildAuthHeaders`, `TASK_COMPLETION_TOOL_NAME`, or `mcpToolNameUtils` in `core-execution` β€” re-export them. They used to be duplicated byte-for-byte across both packages, which was silently load-bearing: if `createToolName` drifted, the tool names `migrate-v16` persisted would stop matching runtime names and every piece/flow/MCP call on a migrated flow would degrade to `ToolCallType.UNKNOWN`. `mcp-tool-name-util.test.ts` asserts both entry points resolve to the *same object*, so a re-fork fails the test rather than shipping. - The four `core/execution/src/lib/agents/` files are **not** uniform. `mcp-tool-name-util.ts` and `mcp.ts` are pure re-export shims (1 and 6 lines). `index.ts` and `tools.ts` re-export the canonical enums and functions but still **own** the execution-side plain-`zod` schema definitions β€” `tools.ts` declares the `AgentTool` union and the `McpAuth*` schemas, `index.ts` declares `AgentOutputField`, `MarkdownContentBlock`, `ToolCallContentBlock` and `AgentStepBlock`. Adding a field to one of those schemas means editing it there *and* in the `zod/mini` twin in `agents.ts`. - **A flow-step run must not reuse chat's resolution logic.** Four separate production failures came from this one assumption while moving the step server-side, each looking like its own bug. `resolveChatProvider` made a step need Chat's provider configured before it would run at all, so an instance that never uses Chat could not run an agent step β€” and it bit twice, because `resolveFastModel` reached the same helper underneath, so every *configured piece tool* failed with a bare `ENTITY_NOT_FOUND` long after the main model had been fixed. Grep for the transitive callers, not just the direct ones. `resolveModelIdForProvider` treats its argument as a *tier* id and falls back to the tier default when it is not in the curated chat list β€” a step configured for `claude-sonnet-4.5` silently ran `4.6`, because a step names a concrete model while chat names a tier. And the chat tool set reaches an unattended run, where a tool that asks the user a question is worse than useless: the agent opened a connection picker, read the empty answer as a refusal, and stopped. When a value crosses between the two surfaces, check what it *means* on each side, not just that the types line up. +- **A worker RPC failure reaches the worker as `error.message` and nothing else.** The envelope in `core/execution/src/lib/engine/rpc.ts` drops `ActivepiecesError.params` and the stack, so three unrelated causes (conversation gone, no chat-enabled provider, pinned provider has no row) all arrive as the same bare `ENTITY_NOT_FOUND` β€” unreadable in the failed-job list. `createRpcServer` logs the intact error on the app side; read *that* log, not the worker's. +- **Whatever enqueues an agent run must pre-check the same thing the worker resolves.** The chat route asked "is any provider enabled for chat" while the worker looked up the run's *pinned* provider, and the flow-step route checked nothing at all β€” so a run enqueued fine and could only fail. Both now call `agentHelpers.assertRunProviderConfigured`, which mirrors the worker's lookup. A pre-check that answers a *different* question than the worker is worse than none: it makes the failure look impossible. +- **Everything the agent job does before its try/catch has no recovery.** `getAgentConfig` used to run outside it, so a config failure sent no error to the chat client and never called `releaseFlowStep` β€” the flow run sat PAUSED until `AP_PAUSED_FLOW_TIMEOUT_DAYS`. Anything added above that block needs its own failure path, or a paused run leaks. - **Build the unattended tool set as an allow-list.** Removing chat tools by name failed three times running β€” display tools, then build-plan and phase tools, then `ap_discover_action_auth` and `ap_load_guide`, which live with the local tools and so survived a filter written by tool group. Grouping tracks where a tool was constructed, not whether it assumes someone is reading. A flow step gets exactly what it is listed: its configured piece actions, the public-web readers, and the structured-output tool. Anything added to chat later stays out by default. - A separate zod-free `agent-primitives.ts` holding those values was tried and **folded back** β€” don't re-create it. It bought no isolation: `core-execution` imports the `@activepieces/core-piece-types` **barrel**, which re-exports `agents.ts`, so `zod/mini` comes along whatever the values live in. - Only the zod *schemas* stay duplicated β€” the `zod` vs `zod/mini` split is a real bundle-size decision, and a schema drift breaks loudly where a function drift did not. diff --git a/brain/knowledge/data-storage-observability/file-storage.md b/brain/knowledge/data-storage-observability/file-storage.md index 6d5d99fd8839..e22efd1093cd 100644 --- a/brain/knowledge/data-storage-observability/file-storage.md +++ b/brain/knowledge/data-storage-observability/file-storage.md @@ -33,8 +33,10 @@ The central service for persisting binary files, backing the execution engine an ### Gotchas - **On cloud the real key prefix is doubled β€” `//…` β€” so ad-hoc CLI work against the bucket silently finds nothing.** Cloud's `AP_S3_ENDPOINT` embeds the bucket as a *path segment* (`https://.r2.cloudflarestorage.com/ap-files-prod`), and `getS3Client` sets `forcePathStyle: true` whenever an endpoint is present, so the SDK appends the bucket again. An object the app stores as `pieces/x.tgz` actually lands at `ap-files-prod/pieces/x.tgz` inside bucket `ap-files-prod`. Symptom when you get it wrong: `aws s3 ls` returns `NoSuchKey` on a *prefix* listing (Aug 2026 β€” cost three attempts to spot). Strip the trailing `/` from the endpoint for the CLI, then prepend the bucket name to the prefix; or better, do bulk work through `s3Helper` so the same client resolves the same paths. Note cloud's object store is **Cloudflare R2** while the piece CDN is a **DigitalOcean Space** β€” two different systems, easy to conflate. +- **The live piece-bundle cache sits *inside* the legacy one β€” `pieces/v2/` is nested under `pieces/`, so a recursive delete of `pieces/` takes the active cache with it.** `S3_PIECES_PREFIX` in `piece-bundle.ts` is `pieces/v2/`; the bare `pieces/` keys beside it are pre-CDN tarballs left by the older writer. Combined with the doubled prefix above, the real keys are `ap-files-prod/pieces/…` (legacy) and `ap-files-prod/pieces/v2/…` (live). Probe both with `wrangler r2 object get` before any prefix-wide operation β€” wiping v2 used to be survivable because it refilled lazily, at the cost of a burst of cache misses on every piece. **Both prefixes are now dead storage and safe to sweep:** the `BUNDLE_PIECE` job and the S3 mirror were removed, so `resolve()` no longer reads or writes either prefix and registry pieces redirect straight to the CDN (else npm). The mirror was deleted because it was written from whichever source was preferred *at cache time* and then took precedence over the CDN forever β€” a bucket populated before the CDN became preferred kept serving the unbundled npm build, which is what fans out one `@activepieces/shared` copy per piece in the engine (see the Workers page). - **`deleteFiles` succeeding does not mean the objects are gone.** `DeleteObjectsCommand` reports per-object failures in `response.Errors` and does **not** throw, and `Quiet: true` only suppresses the success entries β€” so a request that "worked" can still have left objects behind. `deleteFiles` logs a warn naming the failure codes, which is all its callers (best-effort cleanup) need. Anything whose *correctness* depends on the prefix being empty afterwards would have to surface those keys and retry β€” but prefer not to need that at all: a reader that must not see the old objects should read from a new key prefix rather than race a delete against writers that may still be running old code. - Cleanup job runs hourly (`30 */1 * * *`), deletes stale execution files past `EXECUTION_DATA_RETENTION_DAYS`; processes ~4000/iteration, deletes S3 keys in batches of 100. +- **Retention cleanup is row-driven, so any S3 object written without a `file` row is immortal.** The job walks `file` rows and deletes each one's `s3Key`; it never lists the bucket. The piece-tarball cache was exactly that shape β€” keyed by `-.tgz` under a bare prefix, with no row anywhere β€” so nothing has ever swept it and nothing structurally could, whatever the retention setting says. Checked Aug 2026: no migration or job has ever bulk-deleted the `pieces/` prefix, and the only `deleteFiles` callers are the row-driven cleanup and the health probe's own key. If you add a store path that bypasses the `file` table, you own its lifecycle by hand β€” prefer writing a row, or expect a manual `wrangler`/`aws s3` operation forever. - S3 deletes send CRC32C checksum (OCI rejects the SDK-default CRC32). `S3_ENDPOINT` set β†’ SDK checksum/aws-chunked encoding disabled for S3-compatible providers. - `S3_USE_SIGNED_URLS=true` redirects downloads to 7-day pre-signed URLs instead of streaming through the app. diff --git a/brain/knowledge/decisions/000028-piece-tarballs-are-never-mirrored-into-our-own-s3.md b/brain/knowledge/decisions/000028-piece-tarballs-are-never-mirrored-into-our-own-s3.md new file mode 100644 index 000000000000..0c87b9add39a --- /dev/null +++ b/brain/knowledge/decisions/000028-piece-tarballs-are-never-mirrored-into-our-own-s3.md @@ -0,0 +1,25 @@ +--- +status: accepted +--- + +# Piece tarballs are never mirrored into our own S3 + +## Decision +`pieceBundle.resolve()` hands out exactly two kinds of link for a registry piece: the CDN's self-contained bundle when it serves one, otherwise the npm tarball. It does not consult, populate, or fall back to an Activepieces-owned S3 copy, and there is no lazy caching job behind it β€” the `BUNDLE_PIECE` system job was deleted rather than fixed. `ARCHIVE` (custom) pieces are unaffected; they are platform-scoped rows in our own file store and are still streamed from it. + +## Context +Until Aug 2026 `resolve()` checked an S3 mirror first (`pieces/v2/`), and on a miss enqueued `BUNDLE_PIECE` to populate it for next time. That job wrote **whichever source was preferred at cache time** β€” the CDN artifact when `AP_USE_CDN_FOR_BUNDLES` was on, the npm tarball otherwise. + +Because the read order put S3 ahead of the CDN, a bucket populated while the flag was off kept serving the *unbundled npm build* for that piece forever, and turning the flag on later changed nothing for any piece already mirrored. The job carried a comment warning about precisely this, which is the tell that the ordering was known to be load-bearing and fragile. + +The cost is not theoretical. An unbundled tarball declares its build-time dependencies, so each piece pulls its own `@activepieces/shared` and `pieces-framework`. Measured on a 0.88.1 dedicated cloud host: **7 resident `@activepieces/shared` copies holding 208 MB of a 642 MB engine heap** (~33 MB each, ~3,400–4,000 Zod schema objects per copy retained through `require.cache`), 12 distinct versions on disk, and only **7 of 45** installed piece folders carrying a CDN bundle. + +## Why +A cache that can outrank its own upstream is a cache that pins a bug. The mirror's precedence made the *first* fetch permanent, so any later improvement to how pieces are built could not reach a piece that had already been cached β€” and the repair for a poisoned entry is a bucket operation, not a deploy. Deleting the mirror makes the CDN the single upstream, so what a worker installs is decided by the current release rather than by whenever that piece was first requested. + +The obvious alternative was to keep the mirror and only cache CDN artifacts, never npm. Rejected: it keeps a second source of truth whose contents still depend on flag state at write time, still needs a prefix bump every time the meaning of a cached object changes (`pieces/` β†’ `pieces/v2/` was already the second generation), and buys little β€” the CDN is itself a CDN, so we were caching a cache. The mirror's real benefit was egress and independence from npm, and that was never actually delivered, because a cold entry always fetched from npm anyway. + +Not in scope, deliberately: `AP_USE_CDN_FOR_BUNDLES` still defaults to `false`. This decision removes what would *shadow* the CDN once that default flips; flipping it is a separate call. + +## Consequences +Both S3 prefixes (`pieces/`, `pieces/v2/`) become dead storage and can be swept β€” and note `pieces/v2/` is nested inside `pieces/`, so a recursive delete hits both (see *File Storage*). Every piece install now depends on the CDN or npm being reachable, with no local buffer; a self-hoster with S3 configured no longer accumulates a private copy. During a rolling deploy, old instances keep enqueuing `bundle-piece` while new ones have no handler, so those jobs fail with `No handler for job bundle-piece` until the rollout finishes β€” they are cache-warming only, so nothing user-facing breaks. Reintroducing a mirror later means re-deciding the read order, and the rule to keep is that a mirror must never be consulted ahead of the source it was copied from. diff --git a/brain/knowledge/engineering/engineering-handbook-playbooks.md b/brain/knowledge/engineering/engineering-handbook-playbooks.md index d797cc8eae7e..9539672cecbf 100644 --- a/brain/knowledge/engineering/engineering-handbook-playbooks.md +++ b/brain/knowledge/engineering/engineering-handbook-playbooks.md @@ -19,3 +19,6 @@ Run EE, building for self-hosting, setup BetterStack, releases, canary deploymen ## Postmortems & product Postmortems (Redis/queue overload, infra upgrade β€” March 2026) and product interface-design notes. + +## Gotchas +- **`DISCORD_ON_CALL_WEBHOOK` is the single repo secret behind every on-call Discord notification** β€” cloud production deploys (`continuous-delivery-cloud.yml`), self-hosted releases (`release-self-hosted.yml`), and release-pieces failure alerts all post through it. Rotating it repoints all of them at once; there is no per-workflow webhook. Self-hosted release notifications skip `-rc` tags unless `publish_rc_release` is set, mirroring the release-drafter condition so the message never links a release page that doesn't exist. diff --git a/brain/knowledge/engineering/server-module-anatomy.md b/brain/knowledge/engineering/server-module-anatomy.md index 9abab1828333..a20aad6a803d 100644 --- a/brain/knowledge/engineering/server-module-anatomy.md +++ b/brain/knowledge/engineering/server-module-anatomy.md @@ -119,12 +119,16 @@ Register in `app.ts`, in the CE or EE section. EE-only modules live under `src/a Queued work: add to `SystemJobName` or `WorkerJobType` in shared, register the handler via `systemJobHandlers.registerJobHandler()` in `app.ts`. +**Retiring a `SystemJobName` is two steps, and doing only the first orphans jobs forever.** Deleting the enum member removes it from `knownJobNames`, but `isDeprecated()` in `system-job.ts` is `!knownJobNames.includes(name) && deprecatedJobs.some(d => name.startsWith(d))` β€” so a name that is unknown *and* unlisted matches neither branch and is never swept. Whatever is already queued in Redis then survives every `init()`, and `getJobHandler` throws `No handler for job ` on each scan, forever. So also **add the string literal to the `deprecatedJobs` array** in the same file; the 14 names already there are the precedent. Seed one in `test/unit/app/helper/system-jobs/remove-deprecated-jobs.test.ts` β€” its assertions compare the whole remaining queue, so a seeded job is covered for free. + ## Tests `packages/server/api/test/integration/ce/{feature}.test.ts`, using `setupTestEnvironment()` + `createTestContext(app)` β†’ `ctx.post()` / `ctx.get()`. The DB is cleaned between tests. Verify with `npm run lint-dev` and `npm run test-api`. +**`packages/server/api/test/unit/**` runs in no pipeline β€” do not trust it as a safety net.** The package defines a `test-unit` script, but CI only runs `turbo run test-ce test-ee test-cloud check-migrations --filter=api` (`ci.yml`), and the root `npm run test-unit` filters to `engine`/`shared`/`sandbox`/`core-utils`/`server-utils`/`pieces-framework`/`web`/`ee-embed-sdk` β€” `api` is not in that list. So those specs are only ever run by hand, and they rot: measured Aug 2026 on a clean `main`, **18 tests across 4 files already failed** (`workers/job-queue/job-broker`, `workers/machine/machine-service`, `core/canary/worker-group.service`, `knowledge-base/file-service-delete`). Two consequences: put a server test you actually want enforced under `test/integration/ce`, and when a local `test/unit` run goes red, check `main` before assuming your branch caused it. + ## Gotchas - **`getEntities()` and `getMigrations()` are both manual.** Nothing is auto-discovered. A missing entity registration fails silently at runtime; a missing migration registration means the migration simply never runs. diff --git a/brain/knowledge/execution-runtime/workers.md b/brain/knowledge/execution-runtime/workers.md index 21f06179907c..c3445e51707a 100644 --- a/brain/knowledge/execution-runtime/workers.md +++ b/brain/knowledge/execution-runtime/workers.md @@ -28,6 +28,8 @@ The deep `Resolver`/`Runtime` concurrency and bundle-caching model lives on the - **Worker-local periodic work cannot use Redis, `distributedLock`, or a system job β€” it gets a plain `unref`'d timer and must be idempotent.** `AP_CONTAINER_TYPE=WORKER` boots only `packages/server/worker`: no Fastify app, no TypeORM connection, no Redis client, and no Redis env (confirmed in production β€” workers reach the queue only over Socket.IO). `distributedLock` lives in `packages/server/api/src/app/database/redis-connections.ts`, and `worker`'s `package.json` depends on `sandbox`/`server-utils`/`shared`/`core-*` but **not** `api`, and carries neither `ioredis` nor `bullmq` β€” reaching the lock inverts the dependency graph and pulls Fastify, TypeORM and BullMQ into the worker bundle. It is also a trust-boundary change: a worker's only credential is a scoped `AP_WORKER_TOKEN`, which is what lets it run on a machine not trusted with the platform's queue. And a system job is the wrong *shape* regardless of Redis β€” `systemJobsSchedule(...).startWorker()` runs in the app process, once cluster-wide, so it cannot touch a worker's local disk (Helm's default `workloadType: rollout` shares one RWO PVC, but `statefulset` gives each pod its own and the app need not mount it at all). Anything sweeping local state therefore runs on every replica simultaneously by design: make each step idempotent and recompute targets from the live filesystem rather than accumulating state, instead of reaching for a lock. `actionRunCache.sweep` is the worked example β€” see [[action-run]] and decision 000016. - **`SANDBOX_CODE_ONLY` + `concurrency 1` = one immortal engine child allowed to grow to the whole box.** These are two safe-looking settings that only misbehave together. `canReuseSandbox()` (`sandbox-manager.ts`) returns true for `SANDBOX_CODE_ONLY` and `UNSANDBOXED` only, so the engine child is reused forever and `release()` is a no-op β€” the process never resets between jobs. At `concurrency === 1`, `primeFullContainerMemory()` then **overrides the operator's `AP_SANDBOX_MEMORY_LIMIT`** with total container RAM, so that immortal child forks with `--max-old-space-size` = the entire box. V8 feels no pressure and never returns memory. Measured on 0.86.3, arm64, 16 GB box, `AP_SANDBOX_MEMORY_LIMIT=512`: the log reads `fullContainerMemoryKb: 16332416` (the 512 is silently discarded) and the single sandbox RSS climbs 172 β†’ 235 β†’ 241 β†’ 373 β†’ 375 MB across 6.2k runs β€” monotonic, never released. In the process-based modes the sandbox is invalidated after every job, so the same workload shows no growth; that is why this only ever gets reported as a `SANDBOX_CODE_ONLY` "leak". At the default `AP_WORKER_CONCURRENCY=5` priming does not fire, and you instead get **five** persistent children honoring the 512 MB cap (~1.9 GB steady, plateauing) β€” higher floor, but bounded. Note `AP_FLOW_WORKER_CONCURRENCY` does **not** control this; `AP_WORKER_CONCURRENCY` does, and it defaults to `'5'` (`worker/src/lib/config/configs.ts`). - **`cacheState` deliberately holds nothing in memory β€” do not add a memo back.** Disk is the only cache (`cache-state.ts`); every `getOrSetCache` reads `cache.json`. That looks wasteful and is not: measured on a cloud worker, a typical bundle's `cache.json` is 39 KB and costs **0.26 ms** to read, against flow runs measured in seconds. The one case where it is not free is a multi-MB bundle (47 MB β†’ ~220 ms), but `flowBundleStore.tryFetch` already `JSON.parse`s that string twice per run (~123 ms each) via the `cacheMiss` predicate and again on the result, so a memo never saved the dominant cost anyway. It **was** an unbounded `Record` keyed by folder path until Aug 2026, and the failure that caused is the shape to watch for. Unbounded, it was harmless for `pieces-metadata` (5 folders) and fatal for `flow-bundle-store.ts` and `flow-cache.ts`, which put `flowVersionId` **in the path** β€” cardinality became one entry per flow version the worker ever touched, and each bundle value is the entire serialized manifest (flowVersion + pieces + all compiled code), kept as a raw string that is re-parsed on every read anyway. Measured on cloud 0.87.0 before the fix: a worker retained **545 manifests = 170 MB of a ~330 MB heap**, single strings up to 90 MB; the host's shared cache held 19,293 bundle dirs / 1.6 GB (~3 GB as UTF-16), and a shared worker walks all of it because it pulls jobs from every project. Symptom is "memory correlates with pod age". Off-heap stays flat ~100 MB β€” if RSS grows and `Used Heap Size` grows with it, this is the leak, not native/isolated-vm. Retainer path to look for in a snapshot: `Object β†’ property:/…/cache/v13/bundles/ β†’ property: β†’ string`. **Sizing rule for "will tenant X hit this": count bytes, not flows.** The corpus is wildly skewed β€” median bundle **6 KB**, mean 77 KB, p99 1.2 MB, max 47 MB; 95.5% of bundles are under 64 KB and hold 10.5% of the bytes, while the 59 bundles over 4 MB (0.3%) hold 55.8%. So 1,000 median flows is 11 MB and harmless, while ten code-heavy flows is ~400 MB. Model any tenant as `heap β‰ˆ 120 MB baseline + Ξ£(manifest bytes for every distinct flow version executed) Γ— 1.0–1.9` (the Γ—1.9 is V8's two-byte string case, measured: 47.1 MB on disk β†’ 90.3 MB in heap). Note the unit is the flow *version*, so republish churn multiplies it. +- **`AP_REUSE_SANDBOX=true` gives you the immortal engine on `SANDBOX_PROCESS` too, and the memory it holds is `require.cache` β€” bounded, not a leak.** The bullet above reads as if the process-sandboxed modes are safe because `canReuseSandbox()` only returns true for `SANDBOX_CODE_ONLY`/`UNSANDBOXED`; they are not. That function checks `if (!isNil(settings.REUSE_SANDBOX)) return settings.REUSE_SANDBOX === 'true'` **first**, so the env var wins over the mode. Measured Aug 2026 on a 0.88.1 dedicated cloud host (4 Γ— 1 GiB / 0.5 cpu, `SANDBOX_PROCESS`, `concurrency 1`, `AP_SANDBOX_MEMORY_LIMIT` unset, worker at `--max-old-space-size=768`): **memory is a step function of the distinct module set loaded and it converges.** Two engines sampled with a forced GC before each reading β€” a loaded one held 2,615 modules / 9 resident `shared` copies / **517 MB post-GC heap / 668 MB RSS flat across 351 s**, and a fresh one held 8 modules / **46 MB / 183 MB RSS flat across 842 s**. Idle time adds nothing. So "memory correlates with pod age" here is really "correlates with how many distinct piece packages this engine has served". The retainer is `Module._cache`: full paths to the GC root run `zod schema ← property:ShortTextProperty ← property:exports in "Module" ← property: in the 2,531-entry cache object ← property:_cache`, there is no Zod global registry and no per-run accumulation, and the cost is **~3,400–4,000 Zod schema objects per resident copy of `@activepieces/shared`** (plus ~285 per `pieces-framework` copy). Two method notes that cost hours if you get them wrong: attribute by **retainer, not node name** (name-based attribution finds 4.2 MB of a 642 MB heap and points nowhere), and use **post-GC `heapUsed` + module count** as the bounded/unbounded discriminator β€” RSS alone flattens against V8's own `--max-old-space-size` ceiling and proves nothing. +- **The OOM kills are a fit problem, not a growth problem β€” the plateau just doesn't fit the container.** Same host as above: the plateau *height* is set by the piece mix that engine happens to serve β€” 8 modules β†’ 183 MB RSS, 2,615 modules / 9 `shared` copies β†’ 668 MB, 2,532 modules / 7 copies β†’ 807 MB β€” and the heavy end plus a ~150 MB worker plus overhead exceeds a 1 GiB cgroup, so the kernel kills at **827 / 842 / 853 MB anon-rss** while V8 (capped at 1024 MB, larger than the whole container) never feels pressure. Whether a given container dies is luck-of-the-draw on its flows, which is why sibling containers on one host read `oom_kill` 6 / 5 / 3 / 0. **Priced out:** `@activepieces/shared` is **208.2 MB of a 642.6 MB heap across 7 copies** (~33 MB each) plus 8.0 MB for 4 `pieces-framework` copies β€” together 97% of everything attributable to `require.cache` (222.5 MB; the remaining 65% reaches the root via shorter V8-internal paths and so under-attributes modules, making 208 MB a floor). It is mostly **not** the custom pieces: `/root/common` (marketplace) holds **5 copies = 132.5 MB** against `/root/custom_pieces` **2 copies = 75.7 MB**, because every `@europe-express/*` piece pins the same shared while old *marketplace* piece versions each pin a different one (webhookβ†’0.92.0/0.76.7, hubspotβ†’0.87.1/0.37.0, cryptoβ†’0.86.0, subflowsβ†’0.74.0). Collapsing to one resident copy frees ~170 MB and moves the heavy plateaus back under the cgroup β€” the highest-leverage fix, ahead of turning reuse off. Related: `AP_USE_CDN_FOR_BUNDLES` was **unset** on this host and only **7 of 45** installed piece folders carried a `src/bundle.cjs`, which is why the copies exist at all; bundles are not free either, the three largest heap objects were 4.6 MB external source strings held **twice each**. - **Heap ceilings sum inside one cgroup β€” check they fit before blaming a leak.** The worker's `NODE_OPTIONS=--max-old-space-size` and the engine child's `--max-old-space-size` (from `SANDBOX_MEMORY_LIMIT`, default `1048576` KB = 1024 MB in `api/.../system.ts`) are independent ceilings in the *same* container, plus ~90 MB for pm2 and isolate. Cloud prod ran 768 + 1024 + 90 β‰ˆ 1.9 GB of permitted heap in a **1 GiB** container: V8 never felt pressure below 768 MB, so the worker walked past the cgroup and the kernel killed it at ~960 MB anon-rss. Because pm2 restarts it, the container still reads "Up" and `docker stats` looks calm β€” the tell is `State.OOMKilled=true` on a running container (445/448 fleet-wide), a `pm2 list` restart count in the double digits, and `dmesg` lines naming `node /usr/src/a` (the 15-char truncation of the *worker*, not the engine). Every such kill drops an in-flight run. Note `worker.yml` sets `NODE_OPTIONS` per tag but never sets `SANDBOX_MEMORY_LIMIT`, so the engine defaults to a ceiling larger than the whole container. **Measured Aug 2026: fixing the flow-bundle leak did *not* reduce the kill rate at all** β€” 484 containers went from ~1,450 restarts/hr on 0.87.0 to ~1,700/hr at +15 min and ~1,620/hr at +31 min on the fixed 0.88.0, a flat line. Sampled heaps stayed at 128 MB mean / 316 MB max while the kernel kept killing at ~1.01 GB anon-rss, which is the signature of a **fast per-run spike, not accumulation** (a leak raises the sampled mean; this doesn't). So do not assume a proven leak explains the OOMs β€” the over-commit above is the standing suspect and was never corrected. Verify any fix by soaking and comparing restart *rate*, never by looking at a freshly-restarted fleet, which always looks healthy. - **`flowBundleStore.publish` holds three full copies of a flow's compiled code at once, across an RPC await β€” this is what OOM-kills cloud workers.** Confirmed Aug 2026 by snapshotting a worker caught mid-spike: **702 MB of heap against a 768 MB ceiling**, of which 381.3 MB was 16 byte-identical ~23.8 MB `compiledJs` strings and 199.1 MB was `JSArrayBufferData`. `publish` reads every code step concurrently (`Promise.all` over `flowSteps.code`), builds `manifest`, then `Buffer.from(JSON.stringify(manifest))` β€” so the step strings, the serialized string, and the Buffer are all live together. Worse, it then `await`s `prepareFlowBundleUpload`, and the suspended async function's register file pins all three for the whole round trip; the retainer path runs through a 60 s socket-ack `TimersList` to `element:N β†’ property:compiledJs`. One flow with ~16 heavy code steps therefore peaks near 900 MB in a single operation. Note the ordering bug too: the API can answer `skip`, but we build the entire manifest *before* asking. This is a **per-run spike, not a leak** β€” sampled heaps sit at 128 MB mean while kills happen at ~1.01 GB, so a snapshot of a randomly-chosen worker shows nothing; you have to catch one above ~430 MB RSS. Catch it *below* ~600 MB: raising the cgroup does not raise `--max-old-space-size`, so serializing a bigger heap kills the process mid-snapshot (observed). - **Never store process identity in a `cacheState` value β€” the cache directory is shared by every worker container on the host.** `worker.yml` mounts one `/root/cache13` into all ~28 containers, so `cache.json` is cross-process state, not per-process state. `engine-installer` used to write `ENGINE_CACHE_ID = nanoid()` (fresh per process) as the value and treat "value is not mine" as a miss. That only ever worked because the memo fed each process its *own* token back from memory; the moment reads came from disk it saw whichever container wrote last, missed on essentially every job, and re-copied `main.js` fleet-wide β€” production logged `"cacheHit": false` on the engine install for every single job. The rule: a cacheState value must be **content-derived** (a manifest, a compiled artifact, a version) so any process can validate it. Anything that means "did *I* do this?" belongs in a module-level variable, not on shared disk. @@ -37,6 +39,7 @@ The deep `Resolver`/`Runtime` concurrency and bundle-caching model lives on the - **`@activepieces/shared` and `pieces-framework` can never be resident twice β€” that is enforced by the bundler, not by luck.** `bundle-piece-utils.ts` always inlines `@activepieces/*` (the esbuild plugin only ever externalizes third-party packages) and then *filters* `@activepieces/*` out of the declared `external` list before writing the manifest. Verified at runtime on staging Aug 2026 by `Runtime.evaluate` against the live `sandbox-` engine and a heap snapshot: **0 `@activepieces/shared` modules and 0 `pieces-framework` modules** in `require.cache`, on both the worker process (671 modules / 5.22 MB) and the engine (1017 modules / 28.12 MB) β€” and each piece resolves to exactly one module, its `bundle.cjs`. Module keys carry the proof that the *bundled* copy is what loaded: they read `…/pieces/@activepieces/piece-slack-0.9.5/bundle.tgz/…`, reached via `package.json main β†’ src/index.mjs β†’ import './bundle.cjs'`. When auditing this, ignore string-match hits inside a probe's own source text β€” grepping heap strings for the package name finds your own instrumentation first. - **A piece that trips the 5 MB size fallback ships as a "bundle" that still installs its full npm tree, and it dominates engine memory.** `bundlePiece` inlines everything by default, but if the result exceeds `FAIL_BYTES` (5 MB) it rebuilds with `inlineAll: false`, externalizing **all** third-party deps (`bundle-piece-utils.ts`) β€” deliberate, so every piece keeps building and the tarball stays small. The runtime bill lands elsewhere: `google-sheets` has done this since 0.8.4 (10 externals, growing to 17 by 0.16.1), so its bundle declares `googleapis@129.0.0`, which bun installs into the shared workspace store and the engine loads β€” **23.52 MB of a 28.12 MB `require.cache`, 83%, for one piece**, while fully-inlined `slack@0.9.5` contributes 1.58 MB despite being the larger piece. Two `googleapis-common` majors (7.0.1 and 7.2.0) sit resident together, which is exactly the duplication bundling was meant to end. Sampling 60 CDN bundles: 60/60 carry `bundle.cjs`, 55/60 declare **zero** dependencies, and the 5 that don't externalize only native/wasm-ish packages (`pino`, `jimp`, `pdf-lib`, `undici`, `bufferutil`, `tiktoken`). So "bundled" is not a uniform guarantee β€” check `package.json.dependencies` inside the tarball before assuming a piece is self-contained. - **Self-contained piece bundles remove cross-piece dedup, so the per-run spike now scales with how many *fat* pieces one flow touches.** A CDN bundle inlines everything: its `package.json` declares **no dependencies** and every `require()` in `src/bundle.cjs` is a Node built-in. Good for install size and it is what finally kills the multi-copy `@activepieces/shared` problem β€” but two pieces that both use `axios`/`googleapis` no longer share one hoisted copy, they each carry their own. Measured on staging Aug 2026 (`AP_USE_CDN_FOR_BUNDLES=true`, isolate mode): 9 fat bundled pieces in one flow = ~15 MB of uncompressed JS, engine at **926 MB RSS / 788 MB heapUsed** against a V8 `heap_size_limit` of **1216 MB** (`--max-old-space-size=1024` from the default `SANDBOX_MEMORY_LIMIT`) inside a **1 GiB** container already holding a 247 MB worker + 74 MB pm2 β†’ `MEMORY_LIMIT_EXCEEDED`, reproducibly, in ~8 s. The identical flow succeeds with the container raised to 3 GiB. This is the deterministic reproducer for the standing over-commit above: a per-run spike, not accumulation. Sizing rule: budget the engine ceiling against *container minus worker minus pm2*, and treat "many fat pieces in one flow" as the spike driver now that bundles don't share. +- **The cost of a resident `@activepieces/shared` copy is zod schema construction, not data β€” ~40 MB per copy, and reused engines collect one per piece version.** Heap-snapshotted a dedicated-worker engine at 596 MB RSS / 402 MB heap (0.88.1 beta, pre-CDN-bundles, `AP_REUSE_SANDBOX`), Aug 2026: after a double forced GC, 393 MB survived, and the histogram was **2.05 M anonymous closures (109 MB) + 143 MB of `(object properties)` arrays + 359 k `system/Context` scopes (18 MB)** β€” instantiated module graphs, not retained run data (only 16 k distinct functions back those 2 M closures). Every sampled retainer path ended `require.cache β†’ @activepieces+shared@/…/.js β†’ exports β†’ ._def β†’ get shape β†’ refine/pipe/optional/brand closures`: `shared` eagerly builds ~500 top-level zod DTO trees at import, and zod v4 attaches per-instance accessor/method closures (120 k `get`, 67 k `set`, 31 k `validate` in that one heap). The engine held **9 shared versions at once** (0.37 β†’ 0.96.2) because each piece bundle pins its own, and reuse + `import()` pins them forever β€” including **3 versions of `piece-hubspot` and 2 of `piece-slack` simultaneously**, one per flow-pinned piece version, so republish/upgrade churn multiplies copies of the *same* piece. Budget β‰ˆ 60–70 MB engine baseline + ~40 MB per distinct resident shared copy. Two independent fixes attack it: CDN self-contained bundles (no shared inside pieces at all, see above) and require()-based piece LRU eviction; snapshot mechanics for redoing this measurement are in the `profile-worker-memory` skill. Follow-up exact measurement (same process at 542 MB heap, 9 shared copies, graph-cut retained sizes): one cleanly-severable copy weighs **41.3 MB / 445 k nodes**, but the other 8 each show <1 MB *exclusive* retained because copies are **co-retained as clusters, and the pin is the ESM module map, not `require.cache`**: deleting every `require.cache` entry under both `.bun` store roots frees only ~65 MB of 553 MB, while blocking the ESM `ModuleWrap`/`SyntheticModule` entries too frees **406 MB** (engine-only floor: 147 MB, which includes main.js's own bundled shared). Retainer chain: `Global handles β†’ SyntheticModule (piece import()) β†’ Piece object β†’ action run closures β†’ context:pieces_framework_1 / shared_1 (whole exports)` β€” every action closure captures its module scope, so a piece and its framework+shared copies live and die together. Consequence: any eviction scheme must make the *piece entry itself* collectable (hence require()-based loading in the LRU fix β€” an import()ed entry can never be dropped); purging shared's cache entries alone reclaims ~nothing. - **~1.3 GB is the idle floor before a single flow runs.** Same box, freshly booted, zero runs: container 1.325 GiB, of which the API process alone is ~1.18 GB and the worker ~213 MB. Sizing a container off "what a flow needs" is wrong by more than a gigabyte β€” budget the floor first, then add the sandbox ceiling above. - **A new user-interaction `WorkerJobType` must be added to `USER_INTERACTION_JOB_TYPES`** in `packages/server/api/src/app/workers/job-queue/job-queue.ts`. `jobBroker.completeJob` only publishes the engine response back to the waiting webserver for job types in that set β€” miss it and the caller hangs to `WATCHER_SAFETY_TIMEOUT_MS` (5 min) with no error. `submitAndWaitForResponse` has only that backstop, so any *best-effort* caller must additionally cap itself (`Promise.race` with a short timeout); the losing engine job still runs to completion, so the cap buys back user latency, not fleet capacity. - **System-job `No handler` = the worker runs the wrong edition.** The single shared `system-job-queue` is consumed by whichever app instance runs `startWorker()`, and EE handlers only register in the CLOUD/ENTERPRISE branches of the edition switch. A worker on a different edition than the instance that *scheduled* the job throws `No handler for job ` every tick. Seen July 2026: ~14.6k failures, ~99% `chat-stale-sweep`, because the worker defaulted to community (`AP_EDITION` unset) inside a cloud deployment β€” CE jobs like `file-cleanup-trigger` ran fine on the same worker, every EE-scheduled job failed identically. Fix on the deployment (`AP_EDITION=cloud`), not by registering EE handlers in CE. The count looks huge because `removeOnComplete: true` hides successes and `removeOnFail` has an age cap but no count cap. diff --git a/bun.lock b/bun.lock index 207d15bcd621..6249ca9e0e9b 100644 --- a/bun.lock +++ b/bun.lock @@ -117,7 +117,7 @@ }, "packages/core/execution": { "name": "@activepieces/core-execution", - "version": "0.12.0", + "version": "0.13.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", diff --git a/packages/core/execution/package.json b/packages/core/execution/package.json index c3dada522d51..c6257fd9a821 100644 --- a/packages/core/execution/package.json +++ b/packages/core/execution/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/core-execution", - "version": "0.12.0", + "version": "0.13.0", "type": "commonjs", "main": "./dist/src/index.js", "scripts": { diff --git a/packages/core/execution/src/lib/engine/rpc.ts b/packages/core/execution/src/lib/engine/rpc.ts index 1879e417281e..b3e171afa40f 100644 --- a/packages/core/execution/src/lib/engine/rpc.ts +++ b/packages/core/execution/src/lib/engine/rpc.ts @@ -41,6 +41,7 @@ export function createRpcClient( export function createRpcServer( socket: RpcSocket, handlers: T, + log?: RpcLog, ): void { socket.on(RPC_EVENT, async (msg: { method: string, payload: unknown }, ack: (result: unknown) => void) => { const handler = handlers[msg.method as keyof T] @@ -49,6 +50,7 @@ export function createRpcServer( ack(result) } catch (error) { + log?.error({ error, rpc: { method: msg.method } }, 'RPC handler threw') ack({ __rpcError: error instanceof Error ? error.message : String(error) }) } }) @@ -79,3 +81,7 @@ export function createNotifyServer( function isRpcErrorEnvelope(value: unknown): value is { __rpcError: string } { return typeof value === 'object' && value !== null && '__rpcError' in value } + +type RpcLog = { + error(obj: unknown, msg: string): void +} diff --git a/packages/server/api/src/app/ai/ai-provider-service.ts b/packages/server/api/src/app/ai/ai-provider-service.ts index 0ea42cc5943d..a75bb499fe52 100644 --- a/packages/server/api/src/app/ai/ai-provider-service.ts +++ b/packages/server/api/src/app/ai/ai-provider-service.ts @@ -161,6 +161,10 @@ export const aiProviderService = (log: FastifyBaseLogger) => ({ return { provider: chatProvider.provider, auth, config: chatProvider.config, platformId } }, + async exists({ platformId, provider }: { platformId: PlatformId, provider: AIProviderName }): Promise { + return aiProviderRepo().existsBy({ platformId, provider }) + }, + async delete(platformId: PlatformId, providerId: string): Promise { await aiProviderRepo().delete({ platformId, @@ -200,7 +204,7 @@ export const aiProviderService = (log: FastifyBaseLogger) => ({ entityId: provider, entityType: 'AIProvider', }, - }) + }, `the ${provider} AI provider is not configured on this platform`) } let auth = await encryptUtils.decryptObject(aiProvider.auth) diff --git a/packages/server/api/src/app/ee/agent/agent-conversation-controller.ts b/packages/server/api/src/app/ee/agent/agent-conversation-controller.ts index d6e1d8c03b8a..f318c3e4f365 100644 --- a/packages/server/api/src/app/ee/agent/agent-conversation-controller.ts +++ b/packages/server/api/src/app/ee/agent/agent-conversation-controller.ts @@ -4,7 +4,6 @@ import { FastifyBaseLogger } from 'fastify' import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' import { StatusCodes } from 'http-status-codes' import { z } from 'zod' -import { aiProviderService } from '../../ai/ai-provider-service' import { securityAccess } from '../../core/security/authorization/fastify-security' import { assertCreditsAndAppSumoNotExceeded } from '../../platform/billing-provider' import { jobQueue, JobType } from '../../workers/job-queue/job-queue' @@ -150,7 +149,7 @@ export const agentConversationController: FastifyPluginAsyncZod = async (app) => await agentApprovalGate.clearPendingGate({ conversationId }) } - await assertChatProviderConfigured({ platformId, log }) + await agentHelpers.assertRunProviderConfigured({ platformId, log }) await assertCreditsAndAppSumoNotExceeded({ platformId, log }) await jobQueue(runLog).add({ @@ -279,16 +278,6 @@ export const agentConversationController: FastifyPluginAsyncZod = async (app) => } -async function assertChatProviderConfigured({ platformId, log }: { platformId: string, log: FastifyBaseLogger }): Promise { - const provider = await aiProviderService(log).getChatProviderName({ platformId }) - if (isNil(provider)) { - throw new ActivepiecesError({ - code: ErrorCode.ENTITY_NOT_FOUND, - params: { entityId: platformId, entityType: 'ChatAiProvider' }, - }) - } -} - const CHAT_MESSAGES_PER_WINDOW = 40 const CHAT_MESSAGE_RATE_WINDOW_SECONDS = 10 * 60 diff --git a/packages/server/api/src/app/ee/agent/agent-helpers.ts b/packages/server/api/src/app/ee/agent/agent-helpers.ts index d104e2b33965..188b50bdeac4 100644 --- a/packages/server/api/src/app/ee/agent/agent-helpers.ts +++ b/packages/server/api/src/app/ee/agent/agent-helpers.ts @@ -74,11 +74,31 @@ async function resolveChatProvider({ platformId, log }: { platformId: string, lo throw new ActivepiecesError({ code: ErrorCode.ENTITY_NOT_FOUND, params: { entityId: platformId, entityType: 'ChatAiProvider' }, - }) + }, 'no AI provider on this platform is enabled for chat') } return chatProvider } +async function assertRunProviderConfigured({ platformId, provider, log }: { platformId: string, provider?: AIProviderName | null, log: FastifyBaseLogger }): Promise { + if (isNil(provider)) { + const chatProvider = await aiProviderService(log).getChatProviderName({ platformId }) + if (isNil(chatProvider)) { + throw new ActivepiecesError({ + code: ErrorCode.ENTITY_NOT_FOUND, + params: { entityId: platformId, entityType: 'ChatAiProvider' }, + }, 'no AI provider on this platform is enabled for chat') + } + return + } + const configured = await aiProviderService(log).exists({ platformId, provider }) + if (!configured) { + throw new ActivepiecesError({ + code: ErrorCode.ENTITY_NOT_FOUND, + params: { entityId: provider, entityType: 'AIProvider' }, + }, `the ${provider} AI provider is not configured on this platform`) + } +} + function findTier({ tierId }: { tierId: string | null }) { return ACTIVEPIECES_CHAT_TIERS.find((t) => t.id === tierId) } @@ -244,6 +264,7 @@ export const agentHelpers = { getConversationOrThrow, getUserProjects, resolveChatProvider, + assertRunProviderConfigured, resolveTier, resolveModelIdForProvider, resolveModelIdForAnalytics, diff --git a/packages/server/api/src/app/ee/agent/agent-run-controller.ts b/packages/server/api/src/app/ee/agent/agent-run-controller.ts index 72d27f4adb84..0eac2e60aa54 100644 --- a/packages/server/api/src/app/ee/agent/agent-run-controller.ts +++ b/packages/server/api/src/app/ee/agent/agent-run-controller.ts @@ -48,6 +48,7 @@ export const agentRunController: FastifyPluginAsyncZod = async (app) => { }) } const flowTools = await resolveFlowTools({ projectId, flowToolRequests, log: request.log }) + await agentHelpers.assertRunProviderConfigured({ platformId: platform.id, provider, log: request.log }) await assertCreditsAndAppSumoNotExceeded({ platformId: platform.id, log: request.log }) const { ownerId } = await projectService(request.log).getOneOrThrow(projectId) diff --git a/packages/server/api/src/app/helper/system-jobs/common.ts b/packages/server/api/src/app/helper/system-jobs/common.ts index 9b24abcecc8e..099896737053 100644 --- a/packages/server/api/src/app/helper/system-jobs/common.ts +++ b/packages/server/api/src/app/helper/system-jobs/common.ts @@ -14,15 +14,9 @@ export enum SystemJobName { BILLING_USAGE_REPORT = 'billing-usage-report', RESUME_DELAY_WAITPOINT = 'resume-delay-waitpoint', TOOL_SEARCH_REINDEX = 'tool-search-reindex', - BUNDLE_PIECE = 'bundle-piece', CHAT_STALE_SWEEP = 'chat-stale-sweep', } -type BundlePieceSystemJobData = { - name: string - version: string -} - type DeleteFlowDurableSystemJobData = { flow: Flow preDeleteDone: boolean @@ -61,7 +55,6 @@ type SystemJobDataMap = { [SystemJobName.BILLING_USAGE_REPORT]: Record [SystemJobName.RESUME_DELAY_WAITPOINT]: ResumeDelayWaitpointSystemJobData [SystemJobName.TOOL_SEARCH_REINDEX]: ToolSearchReindexSystemJobData - [SystemJobName.BUNDLE_PIECE]: BundlePieceSystemJobData [SystemJobName.CHAT_STALE_SWEEP]: Record } diff --git a/packages/server/api/src/app/helper/system-jobs/system-job.ts b/packages/server/api/src/app/helper/system-jobs/system-job.ts index 21833c9d5183..239ee33ddecb 100644 --- a/packages/server/api/src/app/helper/system-jobs/system-job.ts +++ b/packages/server/api/src/app/helper/system-jobs/system-job.ts @@ -122,6 +122,7 @@ async function removeDeprecatedJobs(log: FastifyBaseLogger): Promise { 'chat-funnel-sync', 'trial-tracker', 'ai-credit-update-check', + 'bundle-piece', ] const knownJobNames = Object.values(SystemJobName) as string[] const isDeprecated = (name: string): boolean => !knownJobNames.includes(name) && deprecatedJobs.some(d => name.startsWith(d)) diff --git a/packages/server/api/src/app/pieces/piece-bundle.ts b/packages/server/api/src/app/pieces/piece-bundle.ts index 65b57c483099..6bed1b876066 100644 --- a/packages/server/api/src/app/pieces/piece-bundle.ts +++ b/packages/server/api/src/app/pieces/piece-bundle.ts @@ -1,20 +1,16 @@ import { isNil, tryCatch } from '@activepieces/core-utils' -import { apDayjs, safeHttp } from '@activepieces/server-utils' +import { safeHttp } from '@activepieces/server-utils' import { FileType, PackageType, PieceType } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { fileRepo } from '../file/file.service' -import { s3Helper } from '../file/s3-helper' import { system } from '../helper/system/system' import { AppSystemProp } from '../helper/system/system-props' -import { SystemJobName } from '../helper/system-jobs/common' -import { systemJobHandlers } from '../helper/system-jobs/job-handlers' -import { systemJobsSchedule } from '../helper/system-jobs/system-job' import { pieceMetadataService } from './metadata/piece-metadata-service' // Resolves a piece to a single downloadable link (see ADR 0002 β€” "Pieces are distributed as links"). -// Official/registry pieces resolve to a signed-S3 object when cached, else to the npm tarball (and a -// lazy SYSTEM job caches it for next time). Custom (ARCHIVE) pieces are served straight from the file -// store. Always platform-scoped via the engine token's platformId. +// Official/registry pieces resolve to the CDN tarball when available, else to the npm tarball. Custom +// (ARCHIVE) pieces are served straight from the file store. Always platform-scoped via the engine +// token's platformId. export const pieceBundle = (log: FastifyBaseLogger) => ({ async resolve({ name, version, archiveId, platformId, projectId }: ResolveParams): Promise { // ARCHIVE pieces are addressed by archiveId β€” they may not be registered in metadata yet @@ -34,16 +30,6 @@ export const pieceBundle = (log: FastifyBaseLogger) => ({ if (metadata.packageType === PackageType.ARCHIVE && !isNil(metadata.archiveId)) { return { type: 'stream', archiveId: metadata.archiveId } } - const s3Enabled = !isNil(system.get(AppSystemProp.S3_BUCKET)) - if (s3Enabled) { - const s3 = s3Helper(log) - const key = pieceBundleS3Key({ name, version }) - if (await s3.objectExists(key)) { - const fileName = `${name.replace('/', '-')}-${version}.tgz` - return { type: 'redirect', url: await s3.getS3SignedUrl(key, fileName) } - } - void tryCatch(() => enqueueBundleJob({ name, version, log })) - } // CDN only mirrors official pieces β€” dev/custom/private registry pieces may 404 there, so fall back to npm. if (metadata.pieceType === PieceType.OFFICIAL && system.getBoolean(AppSystemProp.USE_CDN_FOR_BUNDLES)) { const cdnUrl = cdnTarballUrl({ name, version }) @@ -53,48 +39,8 @@ export const pieceBundle = (log: FastifyBaseLogger) => ({ } return { type: 'redirect', url: npmTarballUrl({ name, version }) } }, - registerJobHandler(): void { - systemJobHandlers.registerJobHandler(SystemJobName.BUNDLE_PIECE, async (data) => { - const s3 = s3Helper(log) - const key = pieceBundleS3Key(data) - if (await s3.objectExists(key)) { - return - } - // The CDN copy is the repackaged, self-contained build; npm may still carry the - // unbundled one. Caching npm here would make S3 β€” which resolve() checks first β€” - // permanently shadow the CDN for this piece. - const source = await preferredTarballSource({ name: data.name, version: data.version, log }) - const response = await safeHttp.retryingAxios.get(source.url, { responseType: 'arraybuffer' }) - await s3.uploadFile(key, Buffer.from(response.data)) - log.info({ - piece: { name: data.name, version: data.version }, - source: source.kind, - }, '[pieceBundle] Cached piece tarball to S3') - }) - }, }) -async function preferredTarballSource({ name, version, log }: PreferredTarballSourceParams): Promise { - if (system.getBoolean(AppSystemProp.USE_CDN_FOR_BUNDLES)) { - const url = cdnTarballUrl({ name, version }) - if (await cdnBundleExists({ url, log })) { - return { kind: 'cdn', url } - } - } - return { kind: 'npm', url: npmTarballUrl({ name, version }) } -} - -async function enqueueBundleJob({ name, version, log }: EnqueueBundleJobParams): Promise { - await systemJobsSchedule(log).upsertJob({ - job: { - name: SystemJobName.BUNDLE_PIECE, - data: { name, version }, - jobId: `bundle-piece:${name}:${version}`, - }, - schedule: { type: 'one-time', date: apDayjs() }, - }) -} - function cdnTarballUrl({ name, version }: PieceRef): string { return `${CDN_PIECES_URL}${name.replace('/', '-')}-${version}.tgz` } @@ -125,37 +71,16 @@ function npmTarballUrl({ name, version }: PieceRef): string { return `${NPM_REGISTRY_URL}/${name}/-/${unscopedName}-${version}.tgz` } -function pieceBundleS3Key({ name, version }: PieceRef): string { - return `${S3_PIECES_PREFIX}${name.replace('/', '-')}-${version}.tgz` -} - const cdnVerifiedUrls = new Set() const NPM_REGISTRY_URL = 'https://registry.npmjs.org' const CDN_PIECES_URL = 'https://cdn.activepieces.com/pieces/bundled/' -// Bumped when what we cache changes meaning. `pieces/` holds tarballs written before the CDN -// became the preferred source, and a rolling deploy keeps writing to it from the old code β€” so -// the new prefix is the only one that can be reached by a writer that prefers the CDN. -const S3_PIECES_PREFIX = 'pieces/v2/' type PieceRef = { name: string version: string } -type EnqueueBundleJobParams = PieceRef & { - log: FastifyBaseLogger -} - -type PreferredTarballSourceParams = PieceRef & { - log: FastifyBaseLogger -} - -type TarballSource = { - kind: 'cdn' | 'npm' - url: string -} - type CdnBundleExistsParams = { url: string log: FastifyBaseLogger diff --git a/packages/server/api/src/app/pieces/piece-sync-service.ts b/packages/server/api/src/app/pieces/piece-sync-service.ts index 90f32a3541a0..2c793041ec3a 100644 --- a/packages/server/api/src/app/pieces/piece-sync-service.ts +++ b/packages/server/api/src/app/pieces/piece-sync-service.ts @@ -14,14 +14,12 @@ import { toolSearchReindexJob } from '../tool-search/tool-search-reindex.job' import { pieceCache } from './metadata/piece-cache' import { PieceMetadataSchema } from './metadata/piece-metadata-entity' import { pieceMetadataService, pieceRepos } from './metadata/piece-metadata-service' -import { pieceBundle } from './piece-bundle' const CLOUD_API_URL = 'https://cloud.activepieces.com/api/v1/pieces' const syncMode = system.get(AppSystemProp.PIECES_SYNC_MODE) export const pieceSyncService = (log: FastifyBaseLogger) => ({ async setup(): Promise { - pieceBundle(log).registerJobHandler() systemJobHandlers.registerJobHandler(SystemJobName.PIECES_SYNC, async function syncPiecesJobHandler(): Promise { await pieceSyncService(log).sync({ publishCacheRefresh: true }) }) diff --git a/packages/server/api/src/app/workers/machine/machine-controller.ts b/packages/server/api/src/app/workers/machine/machine-controller.ts index fe1ada917365..d42ed6fde0ce 100644 --- a/packages/server/api/src/app/workers/machine/machine-controller.ts +++ b/packages/server/api/src/app/workers/machine/machine-controller.ts @@ -21,7 +21,7 @@ export const workerMachineController: FastifyPluginAsyncZod = async (app) => { const assignment = parseWorkerGroupValue({ value: typeof rawWorkerGroupValue === 'string' ? rawWorkerGroupValue : undefined, projectWorker }) const response = await machineService(app.log).onConnection(request, assignment) callback?.(response) - createRpcServer(socket, createHandlers(app.log, assignment, socket.id)) + createRpcServer(socket, createHandlers(app.log, assignment, socket.id), app.log) } }) diff --git a/packages/server/api/test/helpers/mocks/index.ts b/packages/server/api/test/helpers/mocks/index.ts index e5ffb7bc2635..b005165ed011 100644 --- a/packages/server/api/test/helpers/mocks/index.ts +++ b/packages/server/api/test/helpers/mocks/index.ts @@ -679,7 +679,7 @@ export const createMockProjectRelease = (projectRelease?: Partial): Promise> => { +export const createMockAIProvider = async (aiProvider?: Partial & { enabledForChat?: boolean }): Promise> => { return { id: aiProvider?.id ?? apId(), created: aiProvider?.created ?? faker.date.recent().toISOString(), @@ -691,12 +691,12 @@ export const createMockAIProvider = async (aiProvider?: Partial): Pr apiKey: process.env.OPENAI_API_KEY || faker.string.uuid(), }), config: aiProvider?.config ?? {}, - enabledForChat: aiProvider?.provider === AIProviderName.ACTIVEPIECES ? true : false, + enabledForChat: aiProvider?.enabledForChat ?? aiProvider?.provider === AIProviderName.ACTIVEPIECES, } } -export const mockAndSaveAIProvider = async (params?: Partial): Promise> => { +export const mockAndSaveAIProvider = async (params?: Partial & { enabledForChat?: boolean }): Promise> => { const mockAIProvider = await createMockAIProvider(params) await databaseConnection().getRepository('ai_provider').upsert(mockAIProvider, ['platformId', 'provider']) return mockAIProvider diff --git a/packages/server/api/test/integration/ce/pieces/piece-bundle.test.ts b/packages/server/api/test/integration/ce/pieces/piece-bundle.test.ts index bc281f0cf872..7479b008968c 100644 --- a/packages/server/api/test/integration/ce/pieces/piece-bundle.test.ts +++ b/packages/server/api/test/integration/ce/pieces/piece-bundle.test.ts @@ -41,7 +41,7 @@ describe('Piece Bundle Endpoint', () => { expect(response.statusCode).toBe(StatusCodes.UNAUTHORIZED) }) - it('redirects an official piece to the npm tarball when S3 is not configured', async () => { + it('redirects a registry piece to the npm tarball, never to our own S3', async () => { const { mockPlatform, mockProject } = await mockAndSaveBasicSetup() await db.save('piece_metadata', createMockPieceMetadata({ name: '@activepieces/piece-bundle-official', diff --git a/packages/server/api/test/integration/cloud/agent/agent-run-endpoint.test.ts b/packages/server/api/test/integration/cloud/agent/agent-run-endpoint.test.ts index 50be3532eaef..939f2d223196 100644 --- a/packages/server/api/test/integration/cloud/agent/agent-run-endpoint.test.ts +++ b/packages/server/api/test/integration/cloud/agent/agent-run-endpoint.test.ts @@ -1,10 +1,11 @@ -import { apId } from '@activepieces/core-utils' +import { AIProviderName, apId } from '@activepieces/core-utils' import { FastifyInstance } from 'fastify' import { StatusCodes } from 'http-status-codes' import { afterAll, beforeAll, describe, expect, it } from 'vitest' import { accessTokenManager } from '../../../../src/app/authentication/lib/access-token-manager' import { agentHelpers } from '../../../../src/app/ee/agent/agent-helpers' -import { createTestContext } from '../../../helpers/test-context' +import { mockAndSaveAIProvider } from '../../../helpers/mocks' +import { createTestContext, TestContext } from '../../../helpers/test-context' import { setupTestEnvironment, teardownTestEnvironment } from '../../../helpers/test-setup' const RUNS_URL = '/api/v1/agents/runs' @@ -21,7 +22,7 @@ afterAll(async () => { describe('POST /v1/agents/runs', () => { it('writes nothing before publishing, so an interrupted request cannot strand a conversation', async () => { - const ctx = await createTestContext(app) + const ctx = await contextThatCanRunAgents() const engineToken = await accessTokenManager(app.log).generateEngineToken({ jobId: 'job-that-is-not-a-user', projectId: ctx.project.id, @@ -43,7 +44,7 @@ describe('POST /v1/agents/runs', () => { }) it('ignores a project sent in the body and uses the one the engine token is scoped to', async () => { - const ctx = await createTestContext(app) + const ctx = await contextThatCanRunAgents() const other = await createTestContext(app) const engineToken = await accessTokenManager(app.log).generateEngineToken({ jobId: 'job-1', @@ -141,7 +142,7 @@ describe('POST /v1/agents/runs', () => { }) it('accepts the piece tools configured on the step', async () => { - const ctx = await createTestContext(app) + const ctx = await contextThatCanRunAgents() const engineToken = await accessTokenManager(app.log).generateEngineToken({ jobId: 'job-6', projectId: ctx.project.id, @@ -196,7 +197,7 @@ describe('POST /v1/agents/runs', () => { }) it('allows that name when the step has no output fields, since no completion tool is installed', async () => { - const ctx = await createTestContext(app) + const ctx = await contextThatCanRunAgents() const engineToken = await accessTokenManager(app.log).generateEngineToken({ jobId: 'job-8', projectId: ctx.project.id, @@ -240,3 +241,9 @@ describe('POST /v1/agents/runs', () => { expect(response.statusCode).toBe(StatusCodes.BAD_REQUEST) }) }) + +async function contextThatCanRunAgents(): Promise { + const ctx = await createTestContext(app) + await mockAndSaveAIProvider({ platformId: ctx.platform.id, provider: AIProviderName.OPENAI, enabledForChat: true }) + return ctx +} diff --git a/packages/server/api/test/unit/app/helper/system-jobs/remove-deprecated-jobs.test.ts b/packages/server/api/test/unit/app/helper/system-jobs/remove-deprecated-jobs.test.ts index 2759c15cf490..8cf56810610b 100644 --- a/packages/server/api/test/unit/app/helper/system-jobs/remove-deprecated-jobs.test.ts +++ b/packages/server/api/test/unit/app/helper/system-jobs/remove-deprecated-jobs.test.ts @@ -48,6 +48,7 @@ describe('removeDeprecatedJobs', () => { await seedQueue.add('usage-report', {}, { repeat: { pattern: '0 * * * *', tz: 'UTC' } }) await seedQueue.upsertJobScheduler('trial-tracker', { pattern: '0 * * * *', tz: 'UTC' }, { name: 'trial-tracker', data: {} }) await seedQueue.add('issue-reminder', {}, { jobId: 'issue-reminder-one-off', delay: 60_000 }) + await seedQueue.add('bundle-piece', { name: '@activepieces/piece-slack', version: '1.0.0' }, { jobId: 'bundle-piece:@activepieces/piece-slack:1.0.0', delay: 60_000 }) await seedQueue.upsertJobScheduler(SystemJobName.PIECES_ANALYTICS, { pattern: '0 * * * *', tz: 'UTC' }, { name: SystemJobName.PIECES_ANALYTICS, data: {} }) await systemJobsSchedule(log).init() diff --git a/packages/server/worker/src/lib/execute/jobs/ee/agent/agent-mcp-client.ts b/packages/server/worker/src/lib/execute/jobs/ee/agent/agent-mcp-client.ts index 1510ab5843c7..85c0d4dcf8bf 100644 --- a/packages/server/worker/src/lib/execute/jobs/ee/agent/agent-mcp-client.ts +++ b/packages/server/worker/src/lib/execute/jobs/ee/agent/agent-mcp-client.ts @@ -336,7 +336,7 @@ function withToolTimeouts({ mcpToolSet, brokenConnectors, getSelectedAuth, saveL return result } -type McpConnection = { +export type McpConnection = { mcpClient: Awaited> | null mcpToolSet: Record } diff --git a/packages/server/worker/src/lib/execute/jobs/ee/agent/execute-agent-run.ts b/packages/server/worker/src/lib/execute/jobs/ee/agent/execute-agent-run.ts index 295f5474c098..68bf5a01fda1 100644 --- a/packages/server/worker/src/lib/execute/jobs/ee/agent/execute-agent-run.ts +++ b/packages/server/worker/src/lib/execute/jobs/ee/agent/execute-agent-run.ts @@ -3,7 +3,7 @@ import { agentAiUtils } from '@activepieces/server-utils' import { AgentEvent, AgentEventType, AgentKnowledgeBaseTool, AgentMcpTool, AgentOutputField, AgentPhase, AgentPieceTool, AgentResult, AgentRunSource, AgentTool, AgentToolType, EngineResponseStatus, ExecuteAgentRunJobData, PersistedAgentMessage, PersistedAgentPart, PersistedAgentRole, ResolvedAgentFlowTool, WorkerJobType } from '@activepieces/shared' import { createUIMessageStream, generateText, ModelMessage, streamText, ToolSet, toUIMessageStream } from 'ai' import { FireAndForgetJobResult, JobContext, JobHandler, JobResultKind } from '../../../types' -import { agentMcpClient } from './agent-mcp-client' +import { agentMcpClient, McpConnection } from './agent-mcp-client' import { stepResultFrom } from './agent-step-result' import { agentWorkerTools, GateDecision, TaintState } from './agent-worker-tools' import { delayWithJitter, isTransientFailureText, runAgentTurn } from './run-agent-turn' @@ -44,47 +44,7 @@ export const executeAgentRunJob: JobHandler ctx.apiClient.sendAgentEvent({ ...input, runId }), - userId, - conversationId, - log, - }) - - // dryRun (playground): skip MCP and don't execute tools, so the run has no side effects. - const { mcpClient, mcpToolSet } = dryRun - ? { mcpClient: null, mcpToolSet: {} } - : await agentMcpClient.connect({ mcpCredentials: config.mcpCredentials, conversationId, log }) - const configuredPieceTools = (data.tools ?? []).filter(isPieceTool) - const configuredMcpTools = (data.tools ?? []).filter(isMcpTool) - const configuredKnowledgeBaseTools = (data.tools ?? []).filter(isKnowledgeBaseTool) const sendEventWithRetry = ({ event }: { event: AgentEvent }) => retryWithBackoff({ @@ -94,42 +54,11 @@ export const executeAgentRunJob: JobHandler { - log.error({ conversation: { id: conversationId }, maxTurnMs: MAX_TURN_WALL_CLOCK_MS }, 'Chat turn exceeded max wall-clock β€” aborting') - abortController.abort() - }, MAX_TURN_WALL_CLOCK_MS) - - const checkCancelled = async () => { - const { data: response } = await tryCatch(() => ctx.apiClient.executeAgentTool({ - toolName: '__cancel_check', toolInput: { conversationId, runId }, platformId, userId, source, - })) - if (response?.result === true) { - abortController.abort() - } - } - - const cancelCheckInterval = source === AgentRunSource.CHAT - ? setInterval(() => { - checkCancelled().catch(() => {}) - }, 3_000) - : undefined - - // Continuous liveness signal for the entire turn β€” covers long tool/LLM steps and - // approval waits alike, not just gaps between AI-SDK steps. Refreshes connected - // clients' last-chunk clock (empty keepalive chunk) AND the server-side `updated` - // timestamp, so a slow-but-live turn is never reclaimed as stale by either the - // client stale-check or the server's getConversationOrThrow stale-recovery. - const sendHeartbeat = () => { - void tryCatch(() => ctx.apiClient.sendAgentEvent({ - userId, conversationId, runId, - event: { type: AgentEventType.CHUNK, data: [] }, - })) - void tryCatch(() => ctx.apiClient.heartbeatAgentConversation({ conversationId, runId })) - } - const heartbeatInterval = setInterval(sendHeartbeat, HEARTBEAT_INTERVAL_MS) + let source = jobSource ?? AgentRunSource.CHAT + let mcpClient: McpConnection['mcpClient'] = null + let turnWallClockTimer: NodeJS.Timeout | undefined + let cancelCheckInterval: NodeJS.Timeout | undefined + let heartbeatInterval: NodeJS.Timeout | undefined let answer: AgentResult | undefined const structured: { output?: Record } = {} @@ -154,6 +83,86 @@ export const executeAgentRunJob: JobHandler ctx.apiClient.sendAgentEvent({ ...input, runId }), + userId, + conversationId, + log, + }) + + // dryRun (playground): skip MCP and don't execute tools, so the run has no side effects. + const connection: McpConnection = dryRun + ? { mcpClient: null, mcpToolSet: {} } + : await agentMcpClient.connect({ mcpCredentials: config.mcpCredentials, conversationId, log }) + mcpClient = connection.mcpClient + const mcpToolSet = connection.mcpToolSet + + const configuredMcpTools = (data.tools ?? []).filter(isMcpTool) + const configuredKnowledgeBaseTools = (data.tools ?? []).filter(isKnowledgeBaseTool) + + // Absolute backstop: guarantees the turn tears down even if every finer-grained + // signal misses. Routes through the same abortController as user-cancel and the + // idle watchdog, so it lands in the existing cancel-save branch (status β†’ IDLE). + turnWallClockTimer = setTimeout(() => { + log.error({ conversation: { id: conversationId }, maxTurnMs: MAX_TURN_WALL_CLOCK_MS }, 'Chat turn exceeded max wall-clock β€” aborting') + abortController.abort() + }, MAX_TURN_WALL_CLOCK_MS) + + const checkCancelled = async () => { + const { data: response } = await tryCatch(() => ctx.apiClient.executeAgentTool({ + toolName: '__cancel_check', toolInput: { conversationId, runId }, platformId, userId, source, + })) + if (response?.result === true) { + abortController.abort() + } + } + + cancelCheckInterval = source === AgentRunSource.CHAT + ? setInterval(() => { + checkCancelled().catch(() => {}) + }, 3_000) + : undefined + + // Continuous liveness signal for the entire turn β€” covers long tool/LLM steps and + // approval waits alike, not just gaps between AI-SDK steps. Refreshes connected + // clients' last-chunk clock (empty keepalive chunk) AND the server-side `updated` + // timestamp, so a slow-but-live turn is never reclaimed as stale by either the + // client stale-check or the server's getConversationOrThrow stale-recovery. + const sendHeartbeat = () => { + void tryCatch(() => ctx.apiClient.sendAgentEvent({ + userId, conversationId, runId, + event: { type: AgentEventType.CHUNK, data: [] }, + })) + void tryCatch(() => ctx.apiClient.heartbeatAgentConversation({ conversationId, runId })) + } + heartbeatInterval = setInterval(sendHeartbeat, HEARTBEAT_INTERVAL_MS) + const phaseState: { phase: AgentPhase } = { phase: 'discovery' } const taintState: TaintState = { tainted: source === AgentRunSource.FLOW_STEP } diff --git a/packages/server/worker/test/lib/execute/jobs/ee/agent/execute-agent-run-config-failure.test.ts b/packages/server/worker/test/lib/execute/jobs/ee/agent/execute-agent-run-config-failure.test.ts new file mode 100644 index 000000000000..1ba289d9b6fc --- /dev/null +++ b/packages/server/worker/test/lib/execute/jobs/ee/agent/execute-agent-run-config-failure.test.ts @@ -0,0 +1,86 @@ +import { ActivepiecesError, ErrorCode } from '@activepieces/core-utils' +import { AgentEvent, AgentEventType, AgentRunSource, EngineResponseStatus, ExecuteAgentRunJobData, LATEST_JOB_DATA_SCHEMA_VERSION, WorkerJobType } from '@activepieces/shared' +import { describe, expect, it } from 'vitest' +import { executeAgentRunJob } from '../../../../../../src/lib/execute/jobs/ee/agent/execute-agent-run' +import { JobContext } from '../../../../../../src/lib/execute/types' + +const noopLogger = { + child: () => noopLogger, + info: () => undefined, + warn: () => undefined, + error: () => undefined, + debug: () => undefined, + trace: () => undefined, + fatal: () => undefined, +} + +function buildContext(configError?: Error) { + const events: AgentEvent[] = [] + const resumed: { flowRunId: string, waitpointId: string, output: unknown }[] = [] + const apiClient = { + getAgentConfig: () => Promise.reject(configError ?? new ActivepiecesError({ + code: ErrorCode.ENTITY_NOT_FOUND, + params: { entityId: 'OPENAI', entityType: 'AIProvider' }, + })), + resumeFlowStep: (input: { flowRunId: string, waitpointId: string, output: unknown }) => { + resumed.push(input) + return Promise.resolve() + }, + saveAgentMessages: () => Promise.resolve(), + sendAgentEvent: (input: { event: AgentEvent }) => { + events.push(input.event) + return Promise.resolve() + }, + } + const ctx = { apiClient, log: noopLogger } as unknown as JobContext + return { ctx, events, resumed } +} + +function buildJobData(overrides: Partial): ExecuteAgentRunJobData { + return { + schemaVersion: LATEST_JOB_DATA_SCHEMA_VERSION, + jobType: WorkerJobType.EXECUTE_AGENT_RUN, + conversationId: 'conv-1', + runId: 'run-1', + projectId: 'project-1', + platformId: 'platform-1', + userId: 'user-1', + userMessage: 'do it', + modelName: null, + ...overrides, + } +} + +describe('executeAgentRunJob β€” a config failure must not swallow the turn', () => { + it('releases the waiting flow step with the failure instead of leaving the run paused', async () => { + const { ctx, resumed } = buildContext() + const data = buildJobData({ + source: AgentRunSource.FLOW_STEP, + flowRunId: 'flow-run-1', + waitpointId: 'waitpoint-1', + }) + + await expect(executeAgentRunJob.execute(ctx, data)).rejects.toThrow('ENTITY_NOT_FOUND') + + expect(resumed).toHaveLength(1) + expect(resumed[0].waitpointId).toBe('waitpoint-1') + expect(JSON.stringify(resumed[0].output)).toContain('FAILED') + }) + + it('tells the chat client the turn failed instead of leaving it streaming', async () => { + const { ctx, events } = buildContext() + + await expect(executeAgentRunJob.execute(ctx, buildJobData({ source: AgentRunSource.CHAT }))).rejects.toThrow('ENTITY_NOT_FOUND') + + expect(events.map((event) => event.type)).toEqual([AgentEventType.ERROR, AgentEventType.FINISHED]) + }) + + it('completes the job when the platform is out of credits, so it is not retried or paged', async () => { + const { ctx, events } = buildContext(new Error('You have run out of AI credits')) + + const result = await executeAgentRunJob.execute(ctx, buildJobData({ source: AgentRunSource.CHAT })) + + expect(result.status).toBe(EngineResponseStatus.OK) + expect(events.map((event) => event.type)).toEqual([AgentEventType.ERROR, AgentEventType.FINISHED]) + }) +}) diff --git a/packages/web/src/features/authentication/components/auth-landing/turnstile-widget.tsx b/packages/web/src/features/authentication/components/auth-landing/turnstile-widget.tsx index 4a806354716e..98afed94eba4 100644 --- a/packages/web/src/features/authentication/components/auth-landing/turnstile-widget.tsx +++ b/packages/web/src/features/authentication/components/auth-landing/turnstile-widget.tsx @@ -4,6 +4,7 @@ import { t } from 'i18next'; import { useEffect, useRef, useState } from 'react'; import { flagsHooks } from '@/hooks/flags-hooks'; +import { cn } from '@/lib/utils'; const SCRIPT_ID = 'cf-turnstile'; const SCRIPT_SRC = @@ -51,6 +52,19 @@ export function TurnstileWidget({ const container = useRef(null); const widget = useRef(undefined); const [failed, setFailed] = useState(false); + const [shown, setShown] = useState(false); + + useEffect(() => { + const element = container.current; + if (!element) { + return; + } + const observer = new ResizeObserver(() => + setShown(element.offsetHeight > 0), + ); + observer.observe(element); + return () => observer.disconnect(); + }, []); useEffect(() => { if (!siteKey || !container.current) { @@ -72,6 +86,7 @@ export function TurnstileWidget({ widgetId = window.turnstile.render(container.current, { sitekey: siteKey, appearance: 'interaction-only', + theme: 'light', callback: (token: string) => { setFailed(false); onToken(token); @@ -118,7 +133,9 @@ export function TurnstileWidget({ // sign-in cannot proceed and the person needs to know why. return ( <> -
+
+
+
{failed && (

{t(