feat: remote activity workers — run a stage on a separate credential-free machine#35
Merged
Conversation
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Implements wire protocol schemas and types for the remote activity worker system, including ActivityTask, ActivityReport, and all supporting request/ response interfaces. Schemas provide runtime validation for orchestrator-broker and broker-worker communication. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Implement BrokerStore interface with InMemoryBrokerStore backing for task management. Supports create/get/update plus claimNext() which atomically claims the oldest PENDING task for requested stages and marks it ASSIGNED with lease token and incremented attempt. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…adline/presign) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Closes the double-report / late-presign path: a second report() (or presign()) with the same still-valid leaseToken on an already-REPORTED task now throws "task <id> is not ASSIGNED (status=REPORTED)". Adds a test: submit → lease → report (ok) → second report with same token → rejects.toThrow(/ASSIGNED/i). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Pre-existing format violations (line-length, import ordering) left over from earlier tasks; applying biome format --write to clear them before the e2e commit so CI passes clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… idempotency) Implements Task 12: three acceptance tests prove the full integration — A (happy path: remote heavy stage → artifact by ref → core stage, doubled===6), B (worker-crash recovery via broker re-lease, run still completes), D (idempotent job.execute replay); inherent C (suspend/resume assertions in A/B). Also adds examples/end-to-end.ts (runnable main() with drive loop + worker.start/stop), the optional clock param on buildOrchestrator (backward-compatible), and tsconfig includes for the examples directory. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…est D Replace the descriptive comment in test D with live assertions that drain the worker: processOne() returns true once (the single submitted task), then false (no duplicate submit). Also destructures worker from setup(). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… round-trip metrics, zod dep)
- SPEC.md §2: add non-goal bullet documenting that worker-returned artifact
keys embedded in output are not prefix-validated by the proxy; confused-deputy
cross-run reads are a known v1 boundary; prefix-safe artifact-ref channel deferred to Phase 2.
- README.md: add "Security boundary (v1)" section with the same point in 2-3 sentences.
- define-remote-stage.ts checkCompletion: map outcome.customMetrics into a StageMetrics
and return it alongside output (SPEC §9: return metrics).
- define-remote-stage.test.ts: new assertion that a reported-completed poll with
customMetrics: { items: 5 } yields checkCompletion result with metrics.items === 5.
- package.json: promote zod from devDependencies to dependencies (package exports
zod schema values; consumers need zod at runtime).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The example used worker.start() (real-timer polling, 500ms idle) but only looped a fixed 20 attempts (~200ms), so the worker never leased the task in time and the run was abandoned at RUNNING. Now polls fast (idleDelayMs:5) and drives until the run is terminal or a wall-clock timeout. Verified: prints 'Run COMPLETED. seed=3 -> doubled=6'. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ort + e2e - Export ActivityExecutor, ActivityRunInput, ActivityRunResult, BufferedLog, ExecutorDeps from @bratsos/workflow-engine/kernel (Step 0 additive core change) - Create createRemoteExecutor(transport, opts) in host-remote/orchestrator: blocking submit→poll→map model; normalizes worker annotation arg-tuples; maps BufferedProgress items to KernelEvent stage:progress; strict output trust gate via outputSchema.safeParse; carries worker logs through - Export createRemoteExecutor from host-remote/index.ts - Add remote-executor.e2e.test.ts: kernel with RemoteExecutor drives heavyStage and coreStage fully via the port (both registered on worker), asserts run.output.doubled===6 and remote artifact blob present in shared objectStore - TDD: e2e ran RED (module missing), implemented RemoteExecutor, went GREEN - All 30 host-remote tests pass; all 1027 core tests pass; both typechecks clean; lint clean Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Adds createRoutingExecutor to the core kernel: routes specific stage IDs to a remote ActivityExecutor, all others to a lazily-created LocalExecutor (or an explicit local override). Exports createLocalExecutor from kernel/index.ts so consumers can compose it. Proves the mixed model with a host-remote e2e that concretely asserts core ran in-process by verifying the broker store received no task for "core". Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…stamps Review finding: stage:progress KernelEvent timestamps used wall-clock new Date() instead of the injected deps.clock.now(), inconsistent with kernel temporal discipline. The _deps param was unused; now uses deps.clock. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…empotent re-register, version-pin) - Claim-check the worker payload to BlobStore at suspend; recover on broker restart via state:"unknown" -> reload payload -> idempotent re-register. - Preserve the absolute deadline across restarts (set once, never reset). - No-clobber idempotent submit: store.create throws on duplicate; submit returns the existing record without re-creating. Add store.clear() for tests. - Version-pin: a stage-code-version change mid-suspension creates a FAILED task so the run fails instead of resuming on incompatible code. - Broker owns deadline-expiry enforcement (single clock owner); proxy surfaces the broker's specific failure reason via new optional PollResponse.error. - Align broker artifact prefix with the engine stage storage key (rerun-safe). - New restart-recovery.e2e.test.ts (4 tests); no core or fixtures changes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…t recovery Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…rvability, RemoteExecutor taskId+unknown) - Fix 1: null-check payload in checkCompletion "unknown" branch — missing blob returns terminal error instead of silently hanging (permanent suspension) - Fix 2: log WARN in recovery catch so stages stuck in recovery are distinguishable from ones normally waiting - Fix 3: JSDoc on BrokerConfig.stageCodeVersion documenting that version-pinning requires both broker AND worker/proxy to set it; unset broker = pinning disabled - Fix 4: stash deadlineAt + stageCodeVersion in RemotePayload (saved post-submit) and fall back to payload values in recovery if metadata lacks them — self-contained - Fix 5: pass stable taskId (stageRecordId) in RemoteExecutor initial submit for idempotency; handle "unknown" poll state by re-submitting same args and continuing - Fix 6: add comment tying clock.advance(60_001) to fixture's maxWaitMs: 60_000 - Test 1b: new assertion proving Fix 1 — missing payload + broker restart fails the run with "payload missing" error in one poll cycle Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…al cross-machine wire) Adds src/transport/http/ with: - handleBrokerRequest(): pure framework-agnostic handler (unit-testable, no node:http dep) - createBrokerHttpServer(): thin node:http wrapper with ephemeral-port support - createHttpWorkerTransport(): full WorkerTransport over global fetch (Node 22) - /blob endpoint: GET+PUT shim that proxies mem:// presigned URLs over HTTP (v1; swapped for real S3 in Increment C) - Bearer auth on all routes when authToken is configured Exports createBrokerHttpServer, handleBrokerRequest, createHttpWorkerTransport from src/index.ts. 13 unit tests (handleBrokerRequest, no network): auth 401/pass, lease 200/204/409, report 200/409, heartbeat, presign URL shape, blob round-trip, 404 unknown. 3 e2e tests (real TCP, ephemeral loopback port): full workflow COMPLETED doubled===6 with artifact verified in blobStore, auth rejection, auth+e2e pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… on malformed heartbeat/presign body Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…to-bucket artifacts) - Add `createS3Presigner(cfg): ObjectStorePresigner` — uses aws4fetch AwsV4Signer with `signQuery: true` to produce SigV4 presigned PUT/GET URLs for any S3-compatible store (S3, R2, MinIO). TTL is pre-set in the URL query string so aws4fetch picks it up; pathStyle defaults to true for MaxIO/R2/local compat. - Add `createS3BlobStore(cfg): BlobStore` — uses AwsClient.fetch for signed PUT/GET/HEAD/DELETE; LIST uses `?list-type=2&prefix=` and parses `<Key>` XML elements with a regex (no XML lib dependency). - Wire broker-server /presign to return absolute http(s):// URLs as-is when the presigner produces a real S3 URL, and only fall back to the /blob shim for mem:// URLs. Workers write large artifacts directly to the object store with no /blob double-hop. - Export `createS3Presigner`, `createS3BlobStore`, `S3PresignerConfig` from index. - Unit tests (s3-presigner.test.ts): assert URL shape + all SigV4 query params (X-Amz-Algorithm, X-Amz-Credential, X-Amz-Date, X-Amz-Expires, X-Amz-SignedHeaders, X-Amz-Signature) for PUT/GET; virtual-hosted-style; TTL floor; region in credential. - E2E tests (s3-object-store.e2e.test.ts): fake-S3 node:http server (permissive — ignores signature); round-trip BlobStore test (put→get→has→list→delete); full workflow e2e proving worker writes directly to fake-S3 (not /blob) and orchestrator reads via s3BlobStore → run COMPLETED, doubled === 6. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ed URLs, list pagination, XML key decode) - Extract buildObjectUrl to shared url.ts (dedup) - Never send broker Authorization header to presigned/external URLs - Paginate list() via IsTruncated + NextContinuationToken loop - XML-decode entity references in parseListKeys Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…pleted work survives restart) - broker.heartbeat() refreshes leasedAt on every call, preventing false-reap of actively running, heartbeating workers; documents invariant: heartbeatMs must be < staleLeaseMs (default ratio 5 000 ms : 60 000 ms = 1:12) - heartbeats do NOT extend the absolute deadlineAt — confirmed by test - worker writes ActivityReport to <artifactPrefix>/<taskId>/report.json via presigned PUT before calling transport.report(); write is non-fatal so a storage blip never blocks the broker notification - checkCompletion "unknown" branch now checks for the durable report BEFORE re-registering: if found, replays logs/annotations, runs the strict outputSchema trust gate, and returns ready:true without re-running the stage - tests: lease-renewal.e2e.test.ts (3 tests — second worker gets null while first heartbeats; deadline unchanged after heartbeats; stale lease IS reaped without heartbeat); durable-report.e2e.test.ts (2 tests — heavyRunCount===1 asserted before and after recovery poll; fallback re-register when no report) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- examples/cross-process/: spawns worker.ts as a real child process (via pnpm dlx tsx) proving separate-OS-process HTTP round-trip; shared-stages.ts keeps stage IDs/schemas consistent across both files - README.md: full rewrite — in-process + cross-machine quickstarts, executor option comparison, S3/R2 setup, Supported/Limitations section - SPEC.md §2: marks HTTP transport and S3/R2 adapters as DONE; separates genuinely-still-deferred items (mid-cancel, multipart, HA, real-S3 e2e) - .changeset/remote-activity-workers.md: expanded minor changelog - .changeset/workflow-engine-activity-executor.md: minor for core ActivityExecutor port Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…-report, output-key scope, exists 404) - B1: validate durable report with ActivityReportSchema.safeParse; corrupt/forged blobs fall through to re-register; deadline check runs first in unknown branch before any I/O (via injectable _clock for test compat) - B2: broker.report/presign throw on past-deadline; heartbeat returns cancel:true on past-deadline - B3: proxy validates worker output keys against grant prefix; cross-run keys rejected as confused-deputy; artifactPrefix stashed in suspendedState.metadata at execute time - README: fix activityExecutor -> executor in ActivityExecutor port example - scoped-storage: exists() catches not-found errors and returns false - tests: update fixtures/restart-recovery for clock injection; update Test 2b to expect immediate failure (B1 pre-check eliminates second-poll roundtrip) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…rt + object store) Switch the artifact data plane between JSON and binary based on the value type (Uint8Array/ArrayBuffer) on write and the Content-Type header on read. - worker-client: putBytes sends application/octet-stream for Uint8Array/ ArrayBuffer (raw body); getBytes branches on response Content-Type to return Uint8Array or parsed JSON. Auth header still omitted for non-broker presigned URLs (Fix-A). - s3-blob-store: same put/get branching; S3 echoes the Content-Type stored on PUT so binary round-trips correctly. - broker-server: IncomingRequest gains rawBody?: Uint8Array (populated when request Content-Type is application/octet-stream); HandlerResponse gains binary?: Uint8Array (served as octet-stream). PUT /blob stores rawBody for binary, body for JSON; GET /blob returns binary response when the stored value is a Uint8Array, JSON otherwise. - scoped-storage: no changes needed — already passes through to putBytes/ getBytes without any JSON serialization. Adds src/__tests__/binary-artifact.e2e.test.ts: 11 tests covering InMemoryObjectStore, worker-client, S3BlobStore, and full broker-server blob-shim round-trips. Uses bytes [0x00,0x01,0xFF,0xFE,0x80,0x81,0x82] (not valid UTF-8/JSON) as the canonical binary payload. Note: all binary paths buffer the entire body in memory. Streaming support for very large objects (multi-GB ffmpeg output) requires interface changes and is deferred. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…red; presign 409 - B1-gate: add `stageCodeVersion` to `RemoteStageOptions`; in the `"unknown"` durable-report branch, if opts.stageCodeVersion differs from the stashed version in suspendedState.metadata, skip the durable report and fall through to re-register — the broker (configured with the new version) creates a FAILED task, failing the run with a version error rather than completing stale output. - Should-fix 1: `Broker.lease()` now checks the deadline of the claimed task before returning it; an already-expired PENDING task is marked FAILED and null is returned so workers never start uncompletable work. - Should-fix 2: `/presign` in broker-server wraps `broker.presign()` in a try/catch and maps expected broker rejections (fenced lease, deadline, out-of-prefix) to 409 Conflict, consistent with `/lease` and `/report`. - Tests: +4 (deploy-staleness e2e, lease-expiry broker unit, presign-409 handler unit × 2); host-remote 90/90; core 1031/1031; typecheck clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ample + README (enable deploy-safety gate) Codex pre-release note: the durable-report deploy gate is only active when stageCodeVersion is configured on the proxy too, not just broker+workers. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The new package had no changeset, so a release would have shipped it as a junk 0.0.1 (internal-dependency patch bump) with no feature changelog. This makes it an intentional 0.1.0 minor with a proper changelog entry. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ngesets RFC/SPEC/PLAN design docs are no longer tracked (removed from branch history). Only READMEs ship in the repo. Adds .superpowers/ and .claude/ to .gitignore so agent scratch can't be committed again, and removes the two changeset references to the now-untracked RFC. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…skill Extends the existing in-repo skill (the source symlinked into ~/.claude/skills) to cover the new functionality: - SKILL.md: Remote host in the architecture overview, when-to-apply triggers, exports table rows (defineRemoteStage/createActivityWorker, createRoutingExecutor/createLocalExecutor), optional kernel `executor` port, a 'Remote Activity Workers' section, reference-list entry, and a 'Pluggable Execution' principle. Skill version 0.3.0 -> 0.4.0. - references/11-remote-activity-workers.md: full how-to (proxy stage vs ActivityExecutor port, worker SDK, HTTP transport, S3/R2 artifacts, durability, limitations). - migrations/migrate-0.9-to-0.10.md: additive upgrade guide per the skill's maintainer convention (no required actions; new ActivityExecutor port + host-remote package). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Completes the migration chain (0.7->0.8->0.9->0.10) so multi-version upgrades walk cleanly. 0.9 was purely additive (reasoning-model providerOptions passthrough + reasoning-channel access in the AI helper; no schema change), reconstructed from the package CHANGELOG 0.9.0 entry. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What & why
Adds a new package
@bratsos/workflow-engine-host-remoteplus a backward‑compatibleActivityExecutorport in core, so a workflow stage'sexecute()can run on a separate, disposable, credential‑free machine — no database connection, no root object‑store credentials — while a trusted orchestrator owns all state and persistence.Motivating use case: run a heavy stage (e.g. ffmpeg/yt‑dlp transcoding) on a throwaway worker box over the network, streaming large binary artifacts by reference to object storage, restart‑durable. Single trusted orchestrator + a fleet of disposable workers.
How it works
defineRemoteStage) + an in‑process broker (lease / fencing token / deadline / presign) drive the worker through the engine's existing suspend/resume — no kernel change required for this path.createBrokerHttpServer+createHttpWorkerTransport) with bearer‑token auth; workers connect from a separate process/machine.createS3Presigner/createS3BlobStore; S3/R2/MinIO via SigV4): workers PUT large binary artifacts directly via presigned URLs. The broker token is never sent to object storage.SUSPENDEDWorkflowStage+ a claim‑checked payload are the durable anchor; the in‑memory broker rehydrates on poll, with the absolute deadline preserved across restarts.createRoutingExecutor) + a blockingcreateRemoteExecutorfor the in‑coreActivityExecutorport.The core change is backward‑compatible: a default
LocalExecutorreplicates today's in‑process behavior byte‑for‑byte (all 1031 existing core tests unchanged).Proof it actually crosses a process boundary
packages/workflow-engine-host-remote/examples/cross-process/spawns the worker as a separate OS process over HTTP. It leases the heavy stage, writes its artifact via a presigned PUT, and the orchestrator drives the run toCOMPLETED(doubled=8).Verification (this branch)
Release impact (changesets)
Merging this triggers the normal Changesets "Version Packages" PR, which bumps:
@bratsos/workflow-engine-host-remote@bratsos/workflow-engine@bratsos/workflow-engine-host-node@bratsos/workflow-engine-host-serverlessKnown limitations (documented in the package README)
report()is fenced, the run terminates correctly, but the worker isn't interrupted mid‑run.🤖 Generated with Claude Code