feat(ts): add BackgroundTaskEngine - #3838
Conversation
github-actions
Bot
added
area-async
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
|
Assessment: Comment Cleanly-scoped, internal-only slice — no barrel exports, dependencies, or Agent wiring change, so the "no public API" claim holds and no API bar-raising label is required. The engine's lifecycle model (idempotent admission, bounded concurrency, snapshot isolation, execution-scoped cancel/timeout, pause/resume, drain/cancel shutdown) is well-structured and the test suite covers the main paths thoroughly. The feedback below is about hardening one contract and tightening tests/docs, not the design. Review themes
Nice work isolating this from #3632 — the adversarial snapshot/recovery integrity tests are a strong addition. |
7974014 to
3749f93
Compare
3749f93 to
43f0ef7
Compare
|
@strandly-the-agent review please |
strandly-the-agent
left a comment
strandly-the-agent
left a comment
There was a problem hiding this comment.
Verified at 43f0ef7 (branch pr-3838): ✅ npx vitest run --project unit-node src/background-tasks → 22/22 · ✅ eslint / prettier --check / tsc --noEmit --project src/tsconfig.json clean · ✅ 4 repros and 3 test mutations re-run independently rather than relayed · ✅ two suggested fixes applied and re-tested (one lands 22/22 with a 2-line test update; one doesn't work — flagged inline) · ❌ unit-browser not run (no Playwright here) — this file has no env suffix, so CI runs it in Chromium too, and the cloneable set differs there.
🔴 Changes requested. Good slice — the state machine itself held: I couldn't violate either new ! assertion, hang a waiter, leak a listener, or produce an unhandled rejection, and the suite is genuinely strong (whole-object toEqual throughout). The new failure latch is the part that needs another pass: it can't distinguish "the store is broken" from "this one task's data is bad", so one task's result takes down unrelated tasks and leaves shutdown() rejecting forever — including the un-caught shutdown inside #3632's _reloadFromAppState, i.e. the recovery path itself.
- 🔴
engine.ts:495-503— a per-task validation/clone fault latches the whole engine: a healthy sibling's waiter is rejected, its record stuck atworking,cancel/shutdownthrow. #3531 specifies per-task rejection. - 🔴
engine.ts:388— acompletedoutcome with anundefinedresult latches the engine; type-checks clean, and the obviousdelete record.resultfix does not fix it (verified —record.ts:89wants it defined). - 🟡 ×4 — callback sees uncommitted state + re-entrant writes clobbered;
shutdown()sync throw and dropped escalation;requireDatecanonical-only restore; the integrity behaviors' test gap (3 surviving mutations).
Reachability, plainly: nothing imports the engine at merge, so zero user impact today — no 🔴 here can break the shipping SDK. I'm still tiering them 🔴 for this PR rather than "fix in slice 4", because this PR is the review of that design, the consumer already exists (#3632 in-process-task-manager.ts:63-72 instantiates Result from arbitrary tool output; :208 awaits shutdown un-caught), and today's policy is asserted in tests (engine.test.ts:273-299) so it reads as settled once merged. Both fixes are ~10 lines in code this PR introduces. Landing as-is is defensible given the zero reach — but then these two want tracking issues, not silence.
❓ (blocking) onTaskUpdated is synchronous (types.ts:14), while #3531 advertises a pluggable Storage (S3Storage example) that "persists the task before returning the dispatch acknowledgement". How does an async store meet that through a sync hook — does the engine contract change in a later slice, or does durability stay behind appState? Reviewing the engine in isolation is the moment to pin it.
Per-pass breakdown, 3 non-blocking questions, and the non-blocking appendix (18)
Passes. Triage routed 4 of 7: correctness/safety, issue-alignment, adversarial, test-quality. api-bar-raiser / llm-context / docs-accuracy were routed out (internal-only slice, no public API, no model-facing text, no docs). The adversarial pass ran on the default model tier after an advanced-tier infrastructure failure, and the test-quality pass is a time-boxed re-run after a first-attempt timeout — both completed against 43f0ef7, but treat their coverage as slightly shallower than usual.
- correctness/safety — changes requested, 1 blocker (latch scope) + 2 🟡 (
requireDate,shutdownsync throw). 13 items verified OK, including: both new!assertions are statically sound and the two guards removed in the fix commit were unreachable defense-in-depth;_failEnginenever hangs a waiter (waitForIdle/shutdownreject rather than pend); timer bounds exact at 2³¹−1;initializerollback leaves no dirty state; pause/resume identity semantics match the docs; stale timeout timers can't misfire. - issue-alignment — aligned, with under-disclosure. Exactly epic #2496 item 3, nothing outside
background-tasks/(git diff --stat= 5 files, +1855/−0; no dep, barrel,exports, or Agent change), and #3632 already performs the record adaptation this PR promises. But the PR's "the only behavior changes are narrow integrity corrections" doesn't hold: the failure latch is new design that #3531 contradicts, and 5 dropped behaviors + 2 strictness changes go unmentioned. - adversarial — broke-it: 4 landed attacks (1 engine-killing, 3 correctness/contract), 9 held. Held: cancel racing completion, completion after timeout,
resumeaftercancel, executor self-re-entrancy,onEvent('admitted')→cancel, forcing both!assertions, AbortSignal listener leaks (55 waits, 0 leaked), unhandled rejections (4 scenarios), and a seeded 40-task lifecycle fuzz. - test-quality — gaps-found: 6 of the 13 new behaviors are mutation-proof, 1 has no test at all, and the headline behavior (the latch) is half-proven. The earlier review's
objectContainingpoint is fixed — zeroobjectContaining/toMatchObject/toBeTruthyin the file. Isolation is clean (module-levelenginesSet +afterEach, UUID ids, no fake-timer leakage), and the three real-timer budgets are deterministic, not flaky.
Questions (non-blocking)
- Can the body list the 5 behaviors dropped vs #3632 and the 2 strictness changes? (dropped
_executere-check guard,resume's not-found throw,initialize's up-frontclear(), trailing_pump(),submitemitting the live stored object at:117; plusassertTimerDelayon the user-suppliedconfig.timeoutandrequireDate's canonical requirement.) The delta currently reads as purely additive. - Is slice 4 (TaskManager) expected in the same release, so nothing ships unreachable to users? A "unreachable until slice 4" line makes it a knowing merge — I found no prior
strands-tsPR whose entire content was unreachable at merge (closest:sandbox/stream-process.ts, #1090 → consumers in #1110). - Is
wait()resolving onpausedthe intended contract (isSettled,:572-574)? A waiter returns before the task is done and.resultisundefined; if deliberate, an@returnsnote covers it, otherwisewaitForSettledreads truer. Related: shouldDescriptor/Result/Statebe bounded (extends JSONValue) instead of leaving structured-cloneability unexpressed? That's the shared root cause of both 🔴s.
Appendix — non-blocking (18)
Engine behavior: structuredClone strips prototypes/accessors (:109, :569) and Interrupt/InterruptState are classes (src/interrupt.ts:28, :169) — #3632 uses InterruptStateData as State, so an instance type-checks and reaches the resumed executor de-prototyped · a storage failure with no waiter is swallowed and the engine logs nothing (logger.error in _failEngine per strands-ts/AGENTS.md:73-80 would close it) · waitForIdle ignores an already-aborted signal when idle (:536) where wait() throws (:187) · re-entrant submit from onTaskUpdated defeats idempotency (:100-103) · validateStoredEngineTask is unsound for untrusted objects (prototype-chain reads at record.ts:16), safe only because callers clone first · stale snapshots handed out (submit after an observer cancel; executionFinished for a pending-removal task).
Scope / conventions: submit emits the live stored object (:117) where #3632 snapshotted — safe only because _emit clones, and it's the one emit path with no isolation test · failure.type is an open string while the public union is closed, and engine.ts:7 hardcodes 'executionError' · none of the 8 throwing methods carry TSDoc against the @throws rule (strands-ts/AGENTS.md:150); the earlier fix covered interface properties only · MAX_TIMER_DELAY is a new convention, unique in the codebase (swarm.ts:334, graph.ts:325 don't bound delays; graph defaults maxConcurrency to Infinity).
Verification gap: unit-browser unrun — a Buffer in a descriptor/result/state clones in Node but not Chromium, so the same latch would be green in one project and red in the other.
Test nits: updatedAt never proven to advance (:92, mutation survives) · global setTimeout spy fragile and load-bearing, mockRestore outside finally (:624,639) · zero nested describes + two mega-tests (:203, :478) vs docs/TESTING.md:104-136 · toContain('executionFinished') (:474) allows dropped/duplicated events · constructor lower bounds untested (:642 covers only the upper) · waitForStatus microtask-only spin with retry: 2 masking flakes · private-field peeks (:356,932).
No pre-existing bugs to file — every finding is in code this PR introduces.
| private _persistTask(task: StoredEngineTask<Descriptor, Result, State>): void { | ||
| try { | ||
| validateStoredEngineTask(task) | ||
| this._config.onTaskUpdated?.(snapshot(task)) | ||
| } catch (error) { | ||
| this._failEngine(error) | ||
| throw error | ||
| } | ||
| } |
There was a problem hiding this comment.
🔴 _persistTask can't tell "the store is broken" from "this one task's data is bad", so one task's result destroys every other task in the engine.
Verified at 43f0ef7 (unit-node) — task B returns {status:'completed', result:{fn:() => 1}} while unrelated healthy task A is mid-flight:
A's waiter => REJECTED DataCloneError: () => 1 could not be cloned.
A in memory/store => working (forever)
cancel(A) => THROWS DataCloneError
shutdown({cancel}) => THROWS DataCloneError (both modes, permanently — _throwIfFailed at :419)
The engine already gets this right one function up: the same bad value in a descriptor is cloned at :109, outside _persistTask, so submit throws cleanly and the engine keeps working (verified). Only the update path escalates a per-task fault to engine death.
Two consequences I'd rather flag than assert:
- docs: add Background Tasks design #3531 specifies the opposite policy — "If persistence fails, the task is not accepted and the model receives an error instead of a task ID" — per-task rejection, not engine-wide death.
- the next slice can't recover from it:
in-process-task-manager.ts:208(feat(ts): add Background Tasks mechanism #3632) awaitsthis._engine.shutdown({mode:'cancel', …})un-caught inside_reloadFromAppState, the only place that rebuilds a fresh engine — so a latched engine makes the escape hatch throw too.
Validation is per-task and runs before anything is committed (_tasks.set at :463), so it can throw to the caller exactly like submit does, leaving the latch for the genuinely engine-wide case (onTaskUpdated failing):
| private _persistTask(task: StoredEngineTask<Descriptor, Result, State>): void { | |
| try { | |
| validateStoredEngineTask(task) | |
| this._config.onTaskUpdated?.(snapshot(task)) | |
| } catch (error) { | |
| this._failEngine(error) | |
| throw error | |
| } | |
| } | |
| private _persistTask(task: StoredEngineTask<Descriptor, Result, State>): void { | |
| validateStoredEngineTask(task) | |
| try { | |
| this._config.onTaskUpdated?.(snapshot(task)) | |
| } catch (error) { | |
| this._failEngine(error) | |
| throw error | |
| } | |
| } |
Pair that with dropping _failEngine(error) from the clone catch at :456-461 (bare rethrow), and with exempting _shutdownEngine's _throwIfFailed() (:419) so a latched engine can still be closed.
Blast radius, since I ran it: this policy is deliberately asserted today — engine.test.ts:273-299 asserts both the stuck-at-working record and the permanently-rejecting shutdown(). With the change above, exactly that one test fails (21/22), so the test moves with the policy. Which is the real question: is "one bad result → whole engine unrecoverable" the contract you want, or is per-task failure (mirroring the execute-throw handling at :331-340) closer to intent?
| record.status = 'completed' | ||
| delete record.attemptId | ||
| record.result = outcome.result |
There was a problem hiding this comment.
🔴 A task that completes with no result kills the engine — and the obvious fix doesn't work.
:388 writes record.result = outcome.result unguarded. If Result admits undefined (BackgroundTaskEngine<D, void, S> type-checks — the generics are unconstrained), validateCompleted (record.ts:89) throws task.result is required while completed inside _persistTask → _failEngine. Verified at 43f0ef7, vitest type-check clean on the repro:
wait: Error: task.result is required while completed
later submit: Error: task.result is required while completed
shutdown: Error: task.result is required while completed
get(): status "working" <- a task that finished normally, stuck forever
Same shape via resume (:237 → record.ts:69): resume(id, () => ({ state: undefined, ready: true })) → task.state is required while queued for resumption, engine latched.
Don't mirror the failed branch (delete record.result, :379-383): I applied exactly that patch and re-ran — the failure is unchanged, because validateCompleted requires result to be defined and deleting the property leaves it undefined too. The underlying disagreement is inside the PR: types.ts:76-77 documents result?: Result as "Optional execution result retained for completed or failed tasks", while record.ts:89 makes it mandatory.
Pick one:
- resultless completion is legal →
delete record.resulthere and drop therecord.ts:89requirement; completedalways carries a result → bound the generic soResultcannot admitundefined(e.g.Result extends JSONValue, matchingStoredEngineTask's default attypes.ts:59), making this a compile error rather than a runtime engine death.
Either way it should stop being engine-fatal — see the _persistTask comment.
(No caller today, and #3632's Result = ToolResultBlockData doesn't admit undefined — this is a type-legal trap for the next instantiation, e.g. a fire-and-forget task with no payload.)
| this._persistTask(stored) | ||
| this._tasks.set(taskId, stored) |
There was a problem hiding this comment.
🟡 The task map is updated after the persistence hook, so a callback observes a task the engine denies exists — and any write it makes is discarded.
_persistTask (→ onTaskUpdated) runs at :462, this._tasks.set(taskId, stored) at :463; submit has the same order (:115-116). Verified at 43f0ef7: during admission the hook's own record isn't in the map — engine.get(task.taskId) → undefined, engine.cancel(task.taskId) → BackgroundTaskNotFoundError for the task it was just handed. On a terminal record the natural reaction — engine.remove(taskId), which #3632's manager pairs at in-process-task-manager.ts:163-164 from a later call site — throws cannot be removed before reaching a terminal status, because the map still says working. Un-caught inside the hook, that throw becomes engine death.
Second half of the same root cause: _updateTask snapshots at :452 and unconditionally writes that snapshot at :463, so anything the callback chain wrote in between is silently rolled back. Three verified cases: cancel() from onTaskUpdated on the queued→working write returns a cancelled record and the task completes anyway; cancel() from resume's own update callback makes the task execute a second time (runs: 2); shutdown({mode:'cancel'}) from onTaskUpdated waits on a task it believes it cancelled → shutdown timed out, record left working.
The two fixes have to be designed together: simply swapping :462/:463 would defeat the "nothing is committed when validation fails" property the _persistTask comment relies on. The shape that satisfies both is validate → commit → notify: validateStoredEngineTask(stored), then this._tasks.set(taskId, stored), then the onTaskUpdated call (still latching if it throws).
If re-entrancy is out of contract instead, that's a fine answer — it just needs writing down: one line on onTaskUpdated / onEvent (types.ts:13,15) saying a callback must not call back into the engine. Nothing hints at it today and the failure is silent.
| shutdown(options: { readonly mode: 'drain' | 'cancel'; readonly timeout: number }): Promise<void> { | ||
| if (this._shutdown) return this._shutdown | ||
| assertTimerDelay('shutdown timeout', options.timeout) | ||
| this._shutdown = this._shutdownEngine(options).catch((error: unknown) => { | ||
| this._shutdown = undefined | ||
| throw error | ||
| }) | ||
| return this._shutdown | ||
| } |
There was a problem hiding this comment.
🟡 shutdown() validates before a promise exists and memoizes before it looks at options, so cleanup code gets a synchronous throw and an escalation is silently dropped.
Verified at 43f0ef7:
shutdown({mode:'drain', timeout: 0}) -> SYNC THROW TypeError: shutdown timeout must be a positive finite integer, got 0
drain = shutdown({drain, 300}); force = shutdown({cancel, 10})
same promise identity: true
force settled after ~300ms -> Error: Background Task Engine shutdown timed out after 300ms
task after "forceful" shutdown: working <- never cancelled
shutdown(NaN) as a second call: no error <- validation skipped by the memo
The sync throw escapes shutdown(opts).catch(handle) and Promise.allSettled(engines.map((e) => e.shutdown(opts))) — the shape this suite's own afterEach uses (engine.test.ts:31-34). Validating before the memo check and making the method async fixes both halves:
| shutdown(options: { readonly mode: 'drain' | 'cancel'; readonly timeout: number }): Promise<void> { | |
| if (this._shutdown) return this._shutdown | |
| assertTimerDelay('shutdown timeout', options.timeout) | |
| this._shutdown = this._shutdownEngine(options).catch((error: unknown) => { | |
| this._shutdown = undefined | |
| throw error | |
| }) | |
| return this._shutdown | |
| } | |
| async shutdown(options: { readonly mode: 'drain' | 'cancel'; readonly timeout: number }): Promise<void> { | |
| assertTimerDelay('shutdown timeout', options.timeout) | |
| if (this._shutdown) return this._shutdown | |
| this._shutdown = this._shutdownEngine(options).catch((error: unknown) => { | |
| this._shutdown = undefined | |
| throw error | |
| }) | |
| return this._shutdown | |
| } |
Verified: with this change plus a two-line test update (make it('rejects timer delays above the platform limit') async and await expect(...).rejects.toThrow(...) at engine.test.ts:656), the suite is 22/22 and tsc --noEmit is clean.
Left open deliberately: a second call's mode is still ignored, so drain on SIGTERM → cancel on a deadline is a no-op (sequential escalation after a rejection does work — the memo is cleared at :248). Worth either honouring the escalation or documenting that the first call wins.
| requireString(value, path) | ||
| const date = new Date(value) | ||
| if (Number.isNaN(date.getTime()) || date.toISOString() !== value) { | ||
| throw new Error(`${path} must be a canonical ISO-8601 timestamp`) |
There was a problem hiding this comment.
🟡 requireDate accepts only Date.prototype.toISOString() output, which makes restore all-or-nothing against any store that normalizes timestamps.
:142 compares strings, not instants. Verified acceptance set: 2024-01-01T00:00:00.000Z ✅; 2024-01-01T00:00:00Z, 2024-01-01T00:00:00.000+00:00, 2024-01-01T00:00:00.123456Z (Python datetime.isoformat()), 2024-01-01 all ❌. initialize() then throws and rolls back (engine.ts:83-88), so no task is recoverable — every persisted task is lost on restart. Clean error rather than corruption, and #3632 accepted anything Date.parse handled, so this reads as a compatibility decision to state in the PR body rather than a defect.
One correction to the obvious fix, since I tried it: don't relax to Date.parse — Date.parse('08/15/2026') succeeds, Date.parse('2026-02-30T00:00:00.000Z') succeeds by rolling to Mar 2, and …123456Z silently truncates to 123 ms; you'd lose two real rejections that engine.test.ts:853-854 cover. So either keep the strict check and make the contract legible where a store author reads it (types.ts:83,85 say only "Canonical ISO-8601", which doesn't answer "is +00:00 ok?"):
| throw new Error(`${path} must be a canonical ISO-8601 timestamp`) | |
| throw new Error(`${path} must be a canonical ISO-8601 UTC timestamp in 'YYYY-MM-DDTHH:mm:ss.sssZ' form`) |
(verified: 22/22 still green, prettier clean) — or accept ISO-8601 shape variants (offset, extra precision) while still rejecting junk and impossible dates. Cross-SDK parity pushes toward the latter: a Python-written store will emit microseconds (root AGENTS.md:49-54).
| ) | ||
| }) | ||
|
|
||
| it('stops execution when task storage fails', async () => { |
There was a problem hiding this comment.
🟡 The slice's integrity behaviors are its least-tested part — three single-line mutations survive the suite.
Re-verified two myself at 43f0ef7:
- deleting
validateStoredEngineTask(task)from_persistTask(engine.ts:497) → 22/22 still pass. That's "validate on every persist", it runs on every write, and nothing exercises it:engine.test.ts:610,785call the validator directly, which keeps passing. - deleting
if (record.result === undefined) throw …fromvalidateCompleted(record.ts:89) → 22/22 still pass. Same for thefailed/cancelled"required" rules; the table at:788-855covers contradictions but not omissions — one array row each is the cheapest patch in this review. - from the test pass (not re-run here): removing
_throwIfFailed()frominitialize/submit/cancel/wait/resume(engine.ts:43,96,150,185,223) → 22/22. This test proves the latch trips, but never calls a mutating API after it trips, and never asserts the side effects (siblings aborted with the error, queue cleared).
Two drop-in extensions:
// after `await expect(terminalEngine.waitForIdle()).rejects.toBe(terminalError)`
expect(() => terminalEngine.submit({ descriptor: { value: 'again' } })).toThrow(terminalError)
expect(() => terminalEngine.cancel(terminal.taskId, { reason: 'x' })).toThrow(terminalError)
await expect(terminalEngine.wait(terminal.taskId)).rejects.toBe(terminalError)// persist-time validation, currently unproven
const engine = initialize(createEngine(async () => ({ status: 'paused', state: { phase: 'p' } })))
const task = engine.submit({ descriptor: { value: 'work' } })
await engine.wait(task.taskId)
expect(() => engine.resume(task.taskId, () => ({ state: undefined as unknown as TestState, ready: false })))
.toThrow('task.state is required while paused')Two of the review's findings live inside exactly this code, which is why it's worth closing before the next slice depends on it.
43f0ef7 to
6f06b79
Compare
github-actions
Bot
added
complexity/low
|
@strandly-the-agent have another look, thanks |
|
Re-review at Of my four original threads, three are resolved by this commit (property docs incl. the Assessment: Comment (deferring to the standing Changes requested). I concur with @strandly-the-agent's blocker: the new failure latch is the right instinct but wrong scope — it can't separate "store is broken" from "this task's data is bad," so one task's fault stops unrelated tasks and leaves |
3 participants
Description
This PR extracts the first isolated Background Tasks implementation slice planned in #2496: the internal
BackgroundTaskEnginefrom #3632. Reviewing it separately keeps the bounded-execution and lifecycle contract independent from the later TaskManager, plugin, Agent, hook, and tool-execution integration.The engine preserves #3632 behavior for idempotent admission, bounded execution, execution-scoped cancellation and timeouts, pause/resume, persisted-state recovery, observation, and drain/cancel shutdown. This slice changes no dependencies, package exports, public barrels, Agent wiring, or reachable runtime behavior.
Only the generic
StoredEngineTaskdurable record is introduced here. The later TaskManager can specialize and adapt that record into the publicBackgroundTaskview; this split deliberately introduces neitherBackgroundTasknor a second stored-task record type.Relative to #3632, the only behavior changes are narrow integrity corrections found while isolating and adversarially testing the engine: caller-, executor-, and observer-owned values are snapshotted; persisted lifecycle invariants are validated; timestamps must use a valid ISO-8601 shape and calendar date; timer delays are capped at the platform maximum; invalid task data fails that task without poisoning unrelated work; actual persistence-hook failures stop the engine while still allowing shutdown cleanup; observation and shutdown waiters are cleaned up; shutdown options are always validated asynchronously; and initialization after shutdown is rejected. Concurrent shutdown calls retain #3632's first-valid-call-wins mode and deadline rather than adding escalation behavior, and the extraction does not add duplicate restored-record identity policy.
This slice intentionally retains #3632's synchronous, pre-commit
onTaskUpdatedhook because its consumer writes toagent.appStatesynchronously. The async pluggableStoragedescribed in #3531 is not implemented by #3632 and requires a separate persistence contract in a later manager/storage slice.Related Issues
Part of #2496
Parent Background Tasks implementation: #3632
Merged execution-scoped cancellation prerequisite: #3807
Parent feature design (proposed): #3531
Documentation PR
No documentation changes are needed because this split is internal and adds no public API.
Type of Change
New feature
Testing
npx vitest run src/background-tasks/engine/**tests**/engine.test.ts(22 tests in Node and 22 in Chromium)npm run build -w strands-tsnpm run lint -w strands-tsnpm run format:check -w strands-tsnpm run type-check -w strands-tsnpm run complexitynpm run complexity -- --base upstream/main(complexity/low, max 10)git diff --check upstream/mainChecklist
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.