Skip to content

🤖 refactor: convert Config service mutation surface to Effect Semaphore pipeline and config router sites to handlerGen - #4036

Merged
ThomasK33 merged 2 commits into
mainfrom
effect-phase7-config-service
Sep 1, 2026
Merged

ThomasK33 merged 2 commits into
mainfrom
effect-phase7-config-service

Conversation

@ThomasK33

@ThomasK33 ThomasK33 commented Sep 1, 2026 •

Copy link
Copy Markdown
Member

Summary

Phase 7 of the progressive Effect migration (Wave 2): converts the Config service mutation surface to an Effect Semaphore(1)-serialized pipeline, converts the fallible read/write pipelines in ProvidersConfigStore/SecretsStore to Effect per their pre-Effect catch discipline, and moves all 20 config-backed unary router procedures (splashScreens, config, uiLayouts) to handlerGen. All public Promise/sync APIs are preserved by thin runPromise/runSync facades; existing tests pass unchanged.

Background

Follows #4033 (OAuthFlowManager per-flow Scope), #4034 (codex/governor/copilot OAuth + 15 router sites), #4035 (coderOauthService). The Config class serializes all mutations through a private promise-chain queue feeding a private saveConfig — the module docs record the lost-update history (stale-snapshot writes resurrecting removed workspaces). This phase re-expresses that serialization in Effect while preserving those protections exactly.

Implementation

Mutation-surface conversion (src/node/config/index.ts)

  • editConfigQueue promise chain → Semaphore.makeUnsafe(1) (editSemaphore). FIFO permits map the old chain 1:1: each edit's read happens only after the previous edit's write landed, and a failed edit releases its permit on the way out (the old "keep the queue alive on failure" behavior).
  • enqueueConfigEdit stays the only Promise entry point (public editConfig is widely spied/overridden in tests, so named mutators keep routing through it unchanged). It defers the fiber start by one microtask: Effect v4 runPromise executes fibers synchronously until the first async boundary, but the old chain always ran edit bodies on a later microtask, and loadConfigOrDefault's one-shot migrationPersist guard depends on the body observing the guard assignment (otherwise load-time migrations would double-schedule their write-back).
  • The corrupt-file gate (backup-signature CAS) runs inside a single Effect.try step immediately followed by the saveConfig yield — no new await points between the check and the write it approves.
  • saveConfig stays private and keeps its Promise signature (tests spy on it with Promise mocks to simulate swallowed writes); its body moves to saveConfigEffect, an Effect.Effect<void> with a whole-pipeline Effect.catch + Effect.catchDefect fold mirroring the old total try/catch (log-and-swallow; never fails). The serialized pipeline routes the write through the facade so spies keep intercepting it. Load-time-migration write-back semantics (identity transform re-run under the serialized pipeline) are untouched.
  • Raw error identity is preserved end-to-end: v4 runPromise/runSync reject/throw with the original error for both typed failures and defects (verified empirically), so editConfig callers and tests observe byte-identical rejections.

Catch-discipline classification (stores)

Method Pre-Effect discipline Effect shape
Config.saveConfig total try/catch, log-and-swallow whole-pipeline catch + catchDefect fold → Effect<void> (never fails)
Config.enqueueConfigEdit throw-through (gate + transform) Effect.try with catch: (e) => e; raw error to caller via facade
ProvidersConfigStore.loadProvidersConfig total, fold → null (logged) Effect.try + catch fold → Effect<ProvidersConfig | null>
ProvidersConfigStore.getProvidersFileFingerprint total, fold → null (silent) Effect.try + catch fold → Effect<string | null>
ProvidersConfigStore.saveProvidersConfig log-then-rethrow Effect.try + tapError log; raw failure via runSync facade
ProvidersConfigStore.watchProvidersFile callback/watcher lifecycle not converted (fs.watch callback seam; no Effect value)
SecretsStore.loadSecretsConfig / loadRawSecretsConfig total, fold → {} (logged) Effect.try + catch fold
SecretsStore.saveSecretsConfig log-then-rethrow async Effect.tryPromise thunk + tapError; raw rejection via runPromise facade
SecretsStore.updateSecretsBucket (+ updateGlobalSecrets/updateProjectSecrets) throw-through composition Effect.gen composing load → sync bucket fold → save
SecretsStore.getEffectiveSecrets / getGlobalSecrets etc. pure/sync, no catch unchanged (nothing fallible to convert)

Legacy-data passthrough (unsupported secret entries, legacy bestOf metadata, unknown fields) is untouched — the conversion moves control flow only.

fileLeaseManager

Not converted (deliberate). Its lock/lease lifecycle is cross-process (lock directories on disk, PID liveness, stale-breaking, TTLs), not an in-process resource: per the wrap-around-locks doctrine from #4035, Effect wraps around such cross-process critical sections at the caller, never through them. Its Promise seams stay byte-identical.

Router interruption posture (per-procedure)

Reads are single Effect.sync steps (interruption is a don't-care: no partial state possible). Mutations wrap the whole pre-Effect handler body in one Effect.promise thunk, making them uninterruptible by construction: a client abort interrupts the handler fiber but never the in-flight config edit, and multi-step bodies cannot be torn between steps. Rejections become defects → the same internal error the old async handlers produced.

Procedure Kind Posture
splashScreens.getViewedSplashScreens read don't-care (Effect.sync)
splashScreens.markSplashScreenViewed mutation uninterruptible (single thunk)
config.getConfig read don't-care (Effect.sync)
config.onConfigChanged subscription not converted — event-iterator seam waits for the Effect Stream bridge phase
config.updateAgentAiDefaults mutation uninterruptible (single thunk)
config.updateMuxGatewayPrefs mutation ×2 steps uninterruptible (mutate + providerService.notifyConfigChanged in one thunk)
config.updateRoutePreferences mutation (via providerService) uninterruptible (single thunk)
config.updateMinThinkingLevels mutation uninterruptible (single thunk)
config.updateModelFallbacks mutation uninterruptible (single thunk)
config.updateModelPreferences mutation uninterruptible (single thunk)
config.updateCoderPrefs mutation uninterruptible (single thunk)
config.updateRuntimeEnablement mutation uninterruptible (single thunk)
config.saveConfig mutation ×2 steps uninterruptible (saveUserConfig + maybeStartQueuedTasks in one thunk)
config.updateChatTranscriptFullWidth mutation uninterruptible (single thunk)
config.updateLlmDebugLogs mutation uninterruptible (single thunk)
config.updateHeartbeatDefaultPrompt mutation uninterruptible (single thunk)
config.updateHeartbeatDefaultIntervalMs mutation uninterruptible (single thunk)
config.updateGoalDefaults mutation uninterruptible (single thunk)
config.unenrollMuxGovernor mutation ×2 steps uninterruptible (unenroll + policyService.refreshNow in one thunk)
uiLayouts.getAll read don't-care (Effect.sync)
uiLayouts.saveAll mutation uninterruptible (single thunk)

Additionally, enqueueConfigEditEffect itself is Effect.uninterruptible (defense in depth: the corrupt-file gate, write, and change notification form one unit) while waiting for the permit stays interruptible.

Validation

  • Empirical Effect v4 probes (rc.112): runPromise/runSync raw error identity for failures and defects; Semaphore FIFO + serialization; sync fiber start (motivates the microtask defer).
  • src/node/config.test.ts 113/113; src/node/config/ 34/34 (secretsStore, providersConfigStore, fileLeaseManager); workspaceService.configResurrection.test.ts 4/4 (the lost-update contract); router + effectBridge suites; providerService/backup + workspaceService heartbeat/tags/goalDefaults + projectService + updateService + worktreeArchiveSnapshotService consumer suites.
  • tests/ipc/config jest: 7/9 suites green; mcpConfig.test.ts and modelNotFound.test.ts failures reproduce identically on the unmodified base (live-AI bridge environment; see Risks note) — verified via stash/run/pop.
  • Pre-existing env baselines unrelated to this change: taskService (2), workspaceService (bash-monitor-wake), BackupRepoCache 200-commit timeout.

Risks

Severity: medium (config serialization is the app's central persistence chokepoint). The conversion is control-flow-preserving by construction: same read→gate→write→notify order under the same mutual exclusion, same error contracts (verified empirically for identity), same on-disk serialization (untouched). The main behavioral surface is scheduling: the microtask defer preserves the old chain's assignment-before-body ordering that the migration-persist one-shot guard depends on; the config suite pins this (migration write-back tests).

Lessons for Phase 8 (memoryConsolidationService Schedule conversion + workspaceStatusGenerator dispatcher)

  1. Effect v4 fibers start synchronously on runPromise. Any facade replacing a promise-chain/queue must check whether callers depend on deferred body execution (one-shot guards, listener registration windows) and add an explicit microtask defer if so. This will matter for workspaceStatusGenerator's dispatcher loop.
  2. Test-spy seams pin conversion boundaries. editConfig is spied/overridden across many suites, so the Effect pipeline had to live behind the existing method rather than replacing named mutators with Effect surfaces. Check spyOn/property-override usage before choosing the Effect-native surface for memoryConsolidationService.
  3. Wrap-around-locks confirmed again: fileLeaseManager (cross-process dir locks) stayed Promise-native; memoryConsolidationService's file-lock ordering should keep lock interiors byte-identical and put Effect around them.
  4. Merge-queue flake watch: MemoryConsolidationService "rejects a second trigger" 5s-timeout flake evicted 🤖 refactor: convert coderOauthService internals to Effect and adopt handlerGen for coder OAuth procedures #4035 from the queue once; expect it again (verify locally, re-enqueue). Phase 8 touches that very service — consider fixing the flake as part of the phase.
  5. Semaphore.makeUnsafe(1) + withPermits(1) maps AsyncMutex/promise-queue semantics 1:1 (FIFO, release-on-failure), no acquireUseRelease needed.

Generated with xum • Model: anthropic:claude-fable-5 • Thinking: xhigh • Cost: $17.43

@chatgpt-codex-connector

This comment has been minimized.

@ThomasK33

Copy link
Copy Markdown
Member Author

@codex review

@ThomasK33

Copy link
Copy Markdown
Member Author

@codex security review

@chatgpt-codex-connector

This comment has been minimized.

@chatgpt-codex-connector

This comment has been minimized.

@ThomasK33

Copy link
Copy Markdown
Member Author

@codex review

@ThomasK33

Copy link
Copy Markdown
Member Author

@codex security review

@chatgpt-codex-connector

This comment has been minimized.

@chatgpt-codex-connector

This comment has been minimized.

@chatgpt-codex-connector

This comment has been minimized.

@ThomasK33
ThomasK33 added this pull request to the merge queue Sep 1, 2026
Merged via the queue into main with commit bad820b Sep 1, 2026
36 of 38 checks passed
@ThomasK33
ThomasK33 deleted the effect-phase7-config-service branch September 1, 2026 14:30
yermakoffivan pushed a commit to yermakoffivan/mux that referenced this pull request Sep 6, 2026
…erator internals to Effect; make in-flight run-lock reservation deterministic (coder#4038)

## Summary

Phase 8 of the progressive Effect migration (after
coder#4033/coder#4034/coder#4035/coder#4036): converts `memoryConsolidationService` and
`workspaceStatusGenerator` internals to Effect-native pipelines behind
unchanged Promise facades, and fixes the `"rejects a second trigger
while a run is still in flight"` 5s-timeout flake that evicted coder#4035
from the merge queue — by making the in-flight run-lock reservation
deterministic.

## Background

Wave 1 (coder#4031) established the worker templates (Schedule-driven fibers,
runFork sleep fibers); Wave 2 converted the OAuth services and Config.
This phase covers the two remaining Wave 2 services:

- `memoryConsolidationService` (~1100 ln): trigger-driven orchestration
around the dream/harvest runners, with a sidecar `MutexMap` and
per-workspace in-flight promise maps.
- `workspaceStatusGenerator` (~277 ln): the sidebar-status candidate
retry loop with a bounded usage read (previously `Promise.race` +
`setTimeout` — converted to the timeout/sleep-fiber shape from the
idleDispatcher template).

## Flake root cause (chartered fix)

`maybeRun` awaited the durable workspace-removal tombstone
(`fsPromises.access` on the libuv threadpool) **before** the synchronous
in-flight check-and-reserve. Two near-simultaneous triggers both
suspended on that probe, and threadpool completion order — not call
order — decided which caller reserved the run lock. Mutual exclusion
always held (the check-and-reserve itself is one microtask), but the
**winner** was nondeterministic.

In the test, when the _second_ trigger won the reservation, the first
returned the "in flight" refusal and the test then awaited the second
run — which was gated on a model-creation promise the test only releases
_after_ asserting the second call failed. Deadlock → 5s bun timeout.
Same hazard existed in `maybeHarvestThenSweep` (boundary-key coalescing)
and in the sibling `"queues an archive trigger"` test.

**Fix (code, not test):** the funnels now check only the synchronous
in-process teardown mark (`removalCancelled`) before the
check-and-reserve, so reservation order is decided purely by call order
in one synchronous frame; the durable cross-process tombstone probe
moved _behind_ the reservation, to the top of the locked pipelines
(`runLockedEffect` / `harvestThenSweepLockedEffect`), preserving the
same refusal messages and r60/r61 teardown coverage. No awaits were
added between any liveness check and its write. The previously-flaky
tests now pass deterministically (stressed 25×; timeouts unchanged).

## Implementation

**memoryConsolidationService** — Effect.gen pipelines with thin
`Effect.runPromise` facades; wire `Result<_, string>` skip/record unions
stay in the success channel:

- Wrap-around-locks (doctrine from coder#4035/coder#4036): the sidecar `MutexMap`
critical sections remain callback-owned Promise seams with
byte-identical interiors; `saveRecordEffect`/`saveHarvestRecordEffect`
are `Effect.uninterruptible` mutations wrapping those seams. The
`inFlight`/`harvestInFlight` reservation maps remain synchronous
try-lock seams in the funnels (documented in the module header and
`maybeRun` doc) — the funnels stay plain async methods because the
check-and-reserve atomicity is load-bearing.
- `runLockedEffect`, `harvestThenSweepLockedEffect` +
`runHarvestAttemptEffect` (the old try-block),
`recoverRetryableHarvestsEffect`, `runLaunchSweepEffect` (composes
`metaService.effects.getEntries()` directly),
`loadEffect`/`getRecordEffect`/`getStatusEffect`.
- `cancelInFlightConsolidationEffect`: teardown is
`Effect.uninterruptible` end-to-end (r61 mark → abort loop → residual
handoff), with the bounded drain explicitly `Effect.interruptible` +
`Effect.timeout` (it is a wait; the old `Promise.race` + `setTimeout`
timer is gone).
- `triggerInBackground`/`triggerHarvestThenSweepInBackground`: detached
`Effect.runFork` fibers (idleDispatcher template). runFork executes
synchronously up to the first suspension, so trigger-call ordering of
reservations is preserved from the old void-promise chains.

**workspaceStatusGenerator** — `generateWorkspaceStatus` keeps its exact
Promise export (agentStatusService tests spy this module symbol with
`mockResolvedValue`; the spy seam pins the facade). Per-candidate
attempts are one `attemptCandidate` pipeline: the old whole-attempt
try/catch/finally becomes `Effect.catch` + `Effect.catchDefect` folds
(any failure tries the next candidate — no defect escapes the facade
where the old code caught) + `Effect.ensuring` for
`runLanguageModelCleanup`. The 2s usage-read race is now
`Effect.timeout`.

**Interruption posture:** mutations (`saveRecordEffect`,
`saveHarvestRecordEffect`) uninterruptible; teardown
(`cancelInFlightConsolidationEffect`) uninterruptible with an explicitly
interruptible bounded wait; waits (usage read, drain) interruptible so
timeouts work; reads (load/getRecord/getStatus) don't-care. Nothing
externally interrupts these fibers today (all entry points are
runPromise/runFork facades); the posture is for composition safety.

**Catch-discipline audit:** paths uncaught pre-Effect still reject
through facades via `Effect.promise` defects (v4 rethrows raw errors);
paths inside old try/catch blocks are folded (`attemptCandidate`,
`runHarvestAttemptEffect` incl. the completed-record save whose
rejection previously fell into the same catch, `loadEffect` self-healing
with parsing inside the caught thunk, per-iteration `.catch` folds in
recovery/launch-sweep).

**Router sites:** none exist for these services
(`memoryConsolidationService` reaches oRPC only via the `statusChange`
EventEmitter subscription in `routerSubscriptions.ts`, unchanged; the
Effect Stream bridge for subscriptions is Phase 9).

## Validation

- `make static-check` green; `memoryConsolidationService` (40),
`workspaceStatusGenerator` (5), `agentStatusService` (36) and all
`memory*` suites (164) pass unchanged — zero test-file edits.
- Stress: 25 consecutive runs of the two in-flight-race tests + harvest
coalescing test, all green.

## Risks

- Highest-risk area is trigger funnel ordering
(compaction/archive/manual/launch races) and removal teardown (r60/r61).
Mitigations: reservation semantics are strictly tighter (no suspension
before reserve), refusal strings unchanged, lock interiors
byte-identical, and the full behavioral suite passes unchanged.
- Tombstoned-workspace triggers now briefly reserve the run lock before
refusing inside the locked pipeline (previously refused before
reserving). The refusal settles immediately; removal drains observe a
promptly-settling promise, and memory mutations remain gated by the
durable tombstone at commit points.

## Lessons for Phase 9 (Effect Stream bridge, ~24 subscriptions incl.
memory.onChange)

- `memoryConsolidationService` emits `statusChange` + `analyticsIngest`
via EventEmitter; `routerSubscriptions.ts` consumes `statusChange`
through the on/off + `asyncIterableFromSubscription` push seam. The emit
sites are now inside Effect pipelines (`saveRecordEffect` emits after
the lock releases), so bridging to `Stream` can hook those seams
directly — but note emits are _synchronous_ post-write; a Stream bridge
must preserve emit-after-durable-write ordering.
- Fire-and-forget `recordUsage` callbacks (`emit("analyticsIngest")`
from inside provider-stream callbacks) fire from non-Effect contexts (AI
SDK callbacks); the bridge needs a queue that accepts synchronous
emissions from foreign callsites, not just fiber-context offers.
- Spy-seam rule extends to module-level function exports, not just
methods: `agentStatusService.test.ts` spies `generateWorkspaceStatus`
via `spyOn(module, "name")` — module functions consumed through
namespace imports must keep Promise signatures.
- Deterministic-winner lesson generalizes: any funnel whose "who wins"
matters must do check-and-reserve with zero suspensions; converting such
funnels to Effect facades is possible (v4 runs fibers synchronously to
the first suspension) but keeping them as documented synchronous seams
is the honest shape and reviews better.

---

_Generated with `xum` • Model: `anthropic:claude-fable-5` • Thinking:
`xhigh`_

<!-- mux-attribution: model=anthropic:claude-fable-5 thinking=xhigh -->
yermakoffivan pushed a commit to yermakoffivan/mux that referenced this pull request Sep 6, 2026
…coder#4039)

## Summary

Phase 9 of the progressive Effect migration: adds a single reusable
Effect Stream bridge (`src/node/orpc/streamBridge.ts`) for oRPC
subscription procedures and converts all 20 event-subscription handlers
in `routerSubscriptions.ts` to it — including the two deferred by name
from earlier phases (`memory.onChange` from coder#4025 and config
`onConfigChanged` from coder#4036). Wire behavior (payload shapes, ordering,
completion/error semantics) is unchanged; all existing tests pass
unmodified.

## Background

Wave 2 (coder#4033, coder#4034, coder#4035, coder#4036, coder#4038) put every service-backed
unary mutation/query on `handlerGen`/Effect pipelines. What remained on
the router was the subscription surface: ~20 procedures consuming
EventEmitter-based sources via `on`/`off` + the
`asyncIterableFromSubscription` push seam. That seam hand-rolls
lifecycle (abort listeners, queue end, unsubscribe-in-finally) per call.
This phase replaces it on the router with a Scope-managed Effect Stream
pipeline so listener teardown becomes structural instead of
conventional.

## Bridge design (`streamBridge.ts`)

`subscriptionIterable(options)` builds a scoped `Stream` and adapts it
back to the `AsyncGenerator` wire shape oRPC event-iterator procedures
expect:

- **Buffering**: an Effect `Queue` (`Queue.unbounded` for FIFO,
`Queue.sliding(1)` for latest-value coalescing — exact replacement for
`createLatestValueQueue` snapshot semantics).
- **Foreign-callsite emissions**: producers fire from non-Effect
contexts (EventEmitter callbacks, AI-SDK callbacks). `emit.push` is a
synchronous `Queue.offerUnsafe`: the value lands in the buffer before
`push` returns, with **no fiber suspension between the producer's emit
and the queue offer**. This preserves emit-after-durable-write ordering
(e.g. memory consolidation emits `statusChange` synchronously
post-write; subscribers observe events in exactly that order).
- **Guaranteed teardown**: listener attach/detach is wrapped in
`Effect.acquireRelease` (one acquireRelease per resource), tied to the
stream's scope. The scope closes on client disconnect (AbortSignal →
iterator close → fiber interruption), consumer `return()`, stream
failure, and natural completion — `off()`/`removeListener()` runs on
every exit path.
- **Completion**: `emit.end()` maps to `Queue.endUnsafe`, which drains
buffered values before signalling Done (same drain-then-complete
contract the old queues had); `onEnd` then runs and may fail the stream
(bootstrap-error surfacing for `subscribeBackgroundBashes`).
- **Heartbeats**: a scope-tied forked fiber offers the heartbeat value
on an interval, started after `initialize` so heartbeats never
interleave into history replay (replaces `withQueueHeartbeat` on the
router).
- **Ordering of `initial`**: evaluated after attach, delivered before
any buffered events — events firing while a snapshot is computed are
neither lost nor reordered ahead of it.
- **Interruption posture**: subscription streams are interruptible by
design. Aborts interrupt the pull fiber at its next suspension point;
there is no in-flight mutation to protect, only listener handles, which
scope finalizers release. (Deterministic-winner funnels from coder#4038 are
unaffected; none live on this surface.)
- **Defect folding**: validate/subscribe throws and
initialize/initial/onEnd rejections fail the pull and surface as a
rejection of the consumer's `next()` — the same observable contract as
the old seam; nothing escapes as an unhandled rejection.
- **Laziness**: nothing (not even `validate`) runs until the consumer's
first `next()`, matching async-generator semantics.

The wire adapter (`Stream.toAsyncIterable` + an abort-aware generator
wrapper) closes the stream iterator on abort, which interrupts a pending
pull immediately — slightly *prompter* teardown than the old seam (which
detached only when the generator resumed), and wire-invisible.

## Conversion table

| Subscription | Bridge features used | Notes |
|---|---|---|
| `subscribeConfigChanges` | `buffer: "latest"` | deferred by name from
coder#4036 |
| `subscribeMemoryChanges` (`memory.onChange`) | default FIFO; outer
wrapper keeps validate + workspace-identity prelude | deferred by name
from coder#4025 |
| `subscribeProviderConfig`, `subscribePolicyChanges` | `buffer:
"latest"` | |
| `subscribeDevTools` | `initial` (async snapshot) | |
| `subscribeLogs` | `initial` (snapshot captured at attach) | |
| `subscribeTimeline` | `initial` + pre-snapshot event buffering |
catch-up/dedup logic unchanged |
| `subscribeWorkspaceChat` | `heartbeat`, `initialize` (history replay)
| replay relay unchanged; FIFO chunks preserve batch delivery |
| `subscribeMetadata`, `subscribeWorkspaceActivity` | default /
`heartbeat` | |
| `subscribeBackgroundBashes` | `buffer: "latest"`, `initialize`,
`emit.end` + `onEnd` | coalesced reader now constructed at attach (was
eager); bootstrap-error contract unchanged |
| `subscribeWorkspaceStats` | `buffer: "latest"`, `initialize` |
throttle/serialization closure unchanged |
| `subscribeTerminalOutput`, `attachTerminal` | default / `initial`
(screen state) | subscribe-before-capture handshake preserved |
| `subscribeTerminalExit` | `take: 1` | |
| `subscribeTerminalActivity` | `heartbeat`, `initial` | |
| `subscribeUpdateStatus`, `subscribeOpenSettings`,
`subscribeSshPrompts` | default | ssh prompt responder release stays
paired with unsubscribe in one detach thunk |

**Deferred (with reasons):**

| Item | Reason |
|---|---|
| `createTickIterable` (`general.tick`) | Pure timed generator: no event
source to attach, no resource to release — the bridge's acquireRelease
lifecycle adds machinery without value. Documented at the definition. |
| `WorkflowService` internal use of `asyncIterableFromSubscription` |
Service-internal consumption, not a router subscription; out of Phase
9's router-surface scope. The common seam (`asyncEventIterator.ts` etc.)
stays intact for it and for browser-side consumers. |

## Validation

- 12 new behavioral tests in `streamBridge.test.ts` pin the genuinely
new invariants: listener-count-returns-to-baseline after abort (leak
test), teardown on consumer break / stream error / hung-`initialize`
abort / `take` completion (previously unpinned), synchronous-burst emit
ordering, latest-value coalescing, initial-before-buffered ordering,
drain-then-`onEnd`-error completion, heartbeat injection, and laziness.
- `make static-check` green; full `bun test src` green.
- `tests/ipc` run compared against a baseline worktree at the parent
commit in the same environment: failure sets are **identical** (3
suites, all AI-gateway `Forbidden`/PTY environment failures);
`websocketHistoryReplay` and all other subscription-exercising suites
pass with the change.

## Risks

Medium-surface change: every UI subscription (chat, metadata, terminals,
stats, logs, timeline, prompts) now flows through the new bridge. The
main regression classes would be ordering (mitigated: synchronous
`offerUnsafe`, FIFO queues, tests), teardown leaks (mitigated:
per-resource acquireRelease + leak tests), and completion/error
semantics (mitigated: drain-then-Done queue closing, onEnd contract
test, unchanged integration suites). Severity if wrong: stale UI panes
or leaked listeners on long-lived sessions.

## Lessons for Phase 10 (streamManager seams 1–4)

- **Temp-dir Scope (seam 1)**: `Effect.acquireRelease` per resource
composes cleanly under `Stream.unwrap`/scoped effects; the temp-dir
create/cleanup pair should become one acquireRelease rather than
try/finally, and interruption during acquisition is already safe
(release only registers after acquire succeeds).
- **Partial-write debounce fiber (seam 2)**: the scope-tied
`Effect.forkScoped` heartbeat ticker here is the template — a debounce
fiber owned by the stream/pipeline scope gets interrupted with its
owner, removing manual timer bookkeeping. Start such fibers *after*
replay/bootstrap phases if their output must not interleave (same
reasoning as heartbeat-after-initialize).
- **error/lostResponseIds Refs (seam 3)**: this phase kept producer-side
mutable closures (throttle state, bootstrap flags) as plain variables
because producers run in non-Effect contexts. Same rule applies to
streamManager: state mutated from AI-SDK callbacks should stay in plain
mutables or be bridged via `unsafe` APIs (`Queue.offerUnsafe` pattern);
`Ref` is only worth it where fibers are the mutators.
- **Usage accounting (seam 4)**: AI-SDK `recordUsage`-style callbacks
are foreign callsites — the `SubscriptionEmit` pattern (stable function
identities wrapping `unsafe` queue ops, no suspension between callback
and buffer) transfers directly.
- Effect v4 specifics that will recur: `Queue.sliding(1)` is an exact
latest-value-queue replacement; `Queue.endUnsafe` drains before Done
(state `Closing`); `Stream.toAsyncIterable.return()` memoizes its close
promise, so double-close is safe; type the queue's error channel as
`Cause.Done` at creation (`Queue.unbounded<T, Cause.Done>()`) or `end`
won't typecheck.
---

_Generated with `xum` • Model: `anthropic:claude-fable-5` • Thinking:
`xhigh` • Cost: `$15.70`_

<!-- mux-attribution: model=anthropic:claude-fable-5 thinking=xhigh
costs=15.70 -->
yermakoffivan pushed a commit to yermakoffivan/mux that referenced this pull request Sep 6, 2026
…acquireRelease + scope-tied debounce fiber) (coder#4040)

## Summary

Phase 10 (final chartered phase of Wave 2 of the progressive Effect
migration) converts the four incremental resource seams inside
`StreamManager` that Phase 4 (coder#4032) mapped and coder#4039 refined — temp-dir
lifecycle, partial-write debounce, error-categorization state, and usage
accounting — without touching the stream engine core, `StreamManager`'s
public API, or any crash-recovery semantics.

## Background

Phase 4 deliberately deferred wholesale `streamManager.ts` conversion
(5,281 lines; fibers suspend on I/O; AbortController-centric; 30+ field
`WorkspaceStreamInfo`) but identified four seams that convert cleanly.
coder#4039 established the implementation templates this PR applies:
per-resource `Effect.acquireRelease`, scope-tied fibers via
`Effect.forkIn`, the plain-mutables rule for AI-SDK-callback state, and
the zero-suspension foreign-callsite emit pattern.

## Implementation

### Seam 1 — Temp-dir lifecycle → per-resource `Effect.acquireRelease`
(converted)

Each stream now creates a `Scope` (`resourceScope`) in `startStream`.
The temp dir is acquired through `Effect.acquireRelease`: the release
finalizer (the existing fire-and-forget `cleanupStreamTempDir`)
registers only after acquisition succeeds. `Scope.close` replaces the
previous two hand-coordinated cleanup sites (`startStream`'s
`!streamRegistered` finally and `processStreamWithCleanup`'s finally);
close is idempotent, so ownership transfer at registration can never
double-release or leak. The falsy guard for whitebox
`providedRuntimeTempDir: ""` fixtures is preserved inside the finalizer.

### Seam 2 — Partial-write debounce → scope-tied fiber (converted)

`partialWriteTimer` (`setTimeout` handle + three manual `clearTimeout`
sites) becomes `partialWriteFiber`, forked into the stream's
`resourceScope` (coder#4039's scope-tied ticker template, adapted to debounce
semantics). Re-arm and explicit-flush cancellation go through one
`interruptPartialWriteFiber` helper; the stream-teardown cancellation is
now implicit — closing the scope interrupts a pending flush, so a
debounced write can never fire after the stream ends and resurrect
`partial.json` for a dead stream.
`Effect.runSync(Effect.forkIn(...))`/`runFork` execute synchronously up
to the sleep, preserving the previous `setTimeout` registration
ordering; Effect's clock registers a plain `setTimeout` under the hood,
so scheduling semantics are unchanged.

### Seam 3 — Error categorization / `lostResponseIds` (stayed plain,
documented)

Applying coder#4039's plain-mutables rule honestly: `lostResponseIds` is
mutated exclusively from non-Effect contexts (AI-SDK error paths inside
`processStreamWithCleanup`) and read synchronously by `isResponseIdLost`
(a spy-pinned public seam used by `TurnRequestBuilder`). No fiber reads
or mutates it, so a `Ref` would add ceremony without adding safety.
`categorizeError` is pure synchronous classification — nothing to
convert. Both stay plain; the rationale is now documented at the
declaration site. This seam is smaller than the coder#4032 map suggested, by
design.

### Seam 4 — Usage accounting (stayed plain, documented; zero-suspension
invariant pinned in comments)

`cumulativeUsage` / `lastStepUsage` / `cumulativeProviderMetadata` are
mutated only by the AI-SDK `fullStream` loop (`finish-step`) and plain
retry/reset methods — foreign callsites, never fibers — so they stay
plain mutables per the rule. The SubscriptionEmit-transfer property
already holds structurally and is now documented: the `usage-delta` emit
runs with zero suspension after the mutation (`emitTurnEvent` invokes
the sink synchronously), so downstream Effect queue bridges (coder#4039's
`SubscriptionEmit.push` → `Queue.offerUnsafe`) observe usage events in
mutation order.

### Interruption-posture audit

- The debounce fiber is interruptible by design (cancelling a pending
checkpoint write is exactly what `clearTimeout` did). If interrupted
mid-flush, the underlying `flushPartialWrite` promise still runs to
completion — no torn write; writes stay serialized via
`partialWritePromise`.
- Temp-dir release and fiber interruption are synchronous finalizers on
a root fiber (`Effect.runFork(Scope.close(...))`); nothing can interrupt
teardown mid-way.
- No awaits were added between liveness checks and writes; every
abort-signal checkpoint in `startStream` keeps its position.
- Verified empirically against effect@4.0.0-rc.112: `forkIn` into a
closed scope never throws and the fiber never runs; double `Scope.close`
releases exactly once; closing a scope with a sleeping fiber interrupts
it and still runs later finalizers.

### What stayed out (per charter)

The stream engine core (`fullStream` loop, AbortController seams,
`WorkspaceStreamInfo` restructure), `StreamManager`'s public API, and
all spy-pinned seams (`createTempDirForStream` mockResolvedValue spies,
`cleanupStreamTempDir` Reflect-extraction, chaos-test `Reflect.set`
internals) are untouched.

## Validation

- `streamManager.test.ts` (119 existing + 2 new),
`streamManager.chaos.test.ts`,
`streamManager.modelOnlyNotifications.test.ts`, `streamSimulation`,
`replayBufferedStreamMessageRelay`, `agentSession.preStreamError`,
`agentSession.resumeStreamEmptyHistory`, `aiService`, `hooks`,
`turnRequestBuilder` — all pass unchanged.
- New tests cover the two genuinely new invariants: temp-dir release
runs exactly once across the ownership transfer, and a pending debounced
partial write is interrupted at stream end (deterministically armed via
stream-state gating, not sleeps).
- `make static-check` green.
- Standalone probe validated the Effect v4 scope/fiber edge semantics
the design relies on (closed-scope fork, double-close idempotence,
sleeping-fiber interruption).

## Risks

`streamManager.ts` is the most crash-sensitive file in the repo. The
regression surface here is partial-write scheduling and temp-dir
cleanup:

- **Partial-write loss window**: unchanged — a pending debounce
cancelled at teardown was also dropped by the old `clearTimeout`;
terminal paths still perform their own awaited flush/commit.
- **Partial resurrection after stream end**: strictly improved and now
test-pinned (scope close interrupts the pending flush; previously relied
on `clearTimeout` placement).
- **Temp-dir leak/double-free**: strictly improved (idempotent
single-owner release vs. two coordinated call sites + flag).
- Crash-recovery semantics (repairable incomplete state,
malformed-history tolerance) are untouched: no persistence formats, no
request-building paths, no commit predicates changed.

## Wave 2 completion state

Wave 2 (coder#4033 OAuth flow scopes, coder#4034 OAuth services + router sites,
coder#4035 coderOauthService, coder#4036 Config semaphore, coder#4038
consolidation/status generator, coder#4039 oRPC subscription Stream bridge,
this PR) is complete. What remains Promise-based repo-wide, in suggested
Wave 3 order:

1. **StreamManager engine core** — the `fullStream` consumption loop,
AbortController lifecycle, and `WorkspaceStreamInfo` restructure;
deliberately deferred (I/O-suspended fiber teardown needs async
`Scope.close`, which the current sync teardown contract can't host).
Revisit after a ManagedRuntime/Layer DI RFC.
2. **workspace/project/task services** — lock-heavy CRUD services;
natural next targets for the wrap-around-locks +
`Effect.gen`-behind-facade house pattern.
3. **WorkflowService internal event seam** (deferred from coder#4039) and
`createTickIterable` (no resource to manage; convert opportunistically).
4. **ManagedRuntime/Layer dependency injection** — an RFC-scale change
replacing constructor wiring; unlocks `TestClock` for the
timing-sensitive suites and app-lifetime scopes.
5. **Schema at persistence boundaries** — `Schema.decodeUnknown` for
config/history/session artifacts.
6. **OAuth token-refresh worker** (optional; needs product sign-off) and
effect v4 GA + oRPC lockstep upgrades.

---

_Generated with `xum` • Model: `anthropic:claude-fable-5` • Thinking:
`xhigh` • Cost: `$n/a`_

<!-- mux-attribution: model=anthropic:claude-fable-5 thinking=xhigh
costs=n/a -->
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant