feat: bindings persistence, reactor concurrency, thread visibility - #14
Merged
Merged
Conversation
…thread visibility P2: Add persist_binding to cursor_session.ex and opencode_session.ex so resume works for all harness providers, not just Codex. Fix normalize_resume_cursor to re-encode Cursor/OpenCode cursors as JSON strings (their session modules call Jason.decode on the resumeCursor). P0: Add concurrency option to DrainableWorker (default 1, set to 8 in ProviderCommandReactor). Independent thread operations now run in parallel — 12 concurrent session starts no longer timeout sequentially. P1: Stress test reuses bootstrapProjectId from welcome payload instead of creating a separate project, so threads persist in the main sidebar. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughPersist resume bindings for Cursor and OpenCode sessions immediately after emitting "session/ready"; normalize Cursor/OpenCode resume cursors to JSON-encoded strings; make DrainableWorker support configurable concurrency and add tests plus a multi-session resume stress-test script; adjust server session-stop/interrupt handling and domain-event worker concurrency. Changes
Sequence Diagram(s)sequenceDiagram
participant Provider as Provider Session
participant Harness as Harness Runtime
participant Storage as Storage Layer
Provider->>Harness: emit "session/ready"
Harness->>Provider: (ack)
Provider->>Provider: persist_binding(state)
Provider->>Provider: build_resume_cursor() -> cursor_json
Provider->>Storage: upsert_binding(state.thread_id, state.provider, cursor_json)
Storage-->>Provider: persist result
Provider->>Harness: continue with turns
sequenceDiagram
participant EventQueue as Event Queue
participant Reactor as ProviderCommandReactor
participant Worker as DrainableWorker (concurrency: 8)
participant Handler as Event Handler
EventQueue->>Reactor: domain events enqueued
Reactor->>Worker: init worker with concurrency=8
par up to 8 parallel consumers
Worker->>Handler: consumer processes event
Handler-->>Worker: processed
Worker->>Worker: finishOne() tracking
end
Worker-->>Reactor: drained
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
size:XL
vouch:trusted
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
scripts/stress-test-resume-multi.ts (1)
429-430: Hardcoded sleep may cause flakiness.The 5-second sleep after stopping sessions is a heuristic. Under load or on slower systems, sessions may take longer to fully stop, causing race conditions in phase 3.
Consider polling for session stopped state or increasing the timeout for scale mode.
💡 Alternative: poll for session stop confirmation
- // Wait for sessions to fully stop - await sleep(5000); + // Wait for sessions to fully stop + const stopDeadline = Date.now() + 15_000; + while (Date.now() < stopDeadline) { + const allStopped = [...phase1Trackers.values()].every((t) => t.sessionStopped); + if (allStopped) break; + await sleep(500); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@scripts/stress-test-resume-multi.ts` around lines 429 - 430, Replace the hardcoded await sleep(5000) with a deterministic wait that polls each session until it's actually stopped (or a longer configurable timeout) to avoid flakiness; implement or call a helper like waitForSessionsStopped(sessionIds, timeout) or loop using getSessionState(sessionId) / isStopped() with short backoff retries and a max timeout, and use that in place of the sleep in the phase-3 shutdown logic so the test proceeds only after all sessions report stopped.apps/harness/lib/harness/providers/opencode_session.ex (1)
1216-1225: Consider whether storingportin the binding is useful for resume.The stored
portwill be stale if OpenCode restarts or the OS reclaims it. Since resume typically involves spawning a fresh OpenCode server, the persisted port may not be reusable.However, this doesn't cause harm — the port is simply ignored during resume (OpenCode session creates a new server on a new port). If the intent is purely for diagnostics or if there's a future use case for reconnecting to a still-running server, this is fine.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/harness/lib/harness/providers/opencode_session.ex` around lines 1216 - 1225, persist_binding currently writes the ephemeral opencode server port into the stored binding (cursor_json) which will be stale after server restarts and is not used during resume; update the persist_binding function in opencode_session.ex to stop persisting "port" (remove "port" from the JSON) or, if you want to keep it for diagnostics only, add a clear comment and/or a separate diagnostics field so resume logic ignores it; adjust Harness.Storage.upsert_binding call accordingly and ensure any consumers of the stored binding don't rely on the removed "port" value.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@scripts/stress-test-resume-multi.ts`:
- Around line 155-164: The current handler for the "id === 'unknown'" schema
error only rejects the oldest pending promise (inside the block handling id ===
"unknown"), which can leave unrelated in-flight requests unresolved; update that
block to iterate over this.pending and reject every pending entry with a
descriptive Error(`Schema error: ${errMsg}`), then clear the map (or call
this.pending.clear()) so no promises remain outstanding; alternatively, if you
prefer to keep the heuristic, add a clear comment above the existing loop
explaining the limitation and why only the oldest request is rejected (but
prefer rejecting all in this stress-test script).
---
Nitpick comments:
In `@apps/harness/lib/harness/providers/opencode_session.ex`:
- Around line 1216-1225: persist_binding currently writes the ephemeral opencode
server port into the stored binding (cursor_json) which will be stale after
server restarts and is not used during resume; update the persist_binding
function in opencode_session.ex to stop persisting "port" (remove "port" from
the JSON) or, if you want to keep it for diagnostics only, add a clear comment
and/or a separate diagnostics field so resume logic ignores it; adjust
Harness.Storage.upsert_binding call accordingly and ensure any consumers of the
stored binding don't rely on the removed "port" value.
In `@scripts/stress-test-resume-multi.ts`:
- Around line 429-430: Replace the hardcoded await sleep(5000) with a
deterministic wait that polls each session until it's actually stopped (or a
longer configurable timeout) to avoid flakiness; implement or call a helper like
waitForSessionsStopped(sessionIds, timeout) or loop using
getSessionState(sessionId) / isStopped() with short backoff retries and a max
timeout, and use that in place of the sleep in the phase-3 shutdown logic so the
test proceeds only after all sessions report stopped.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: e82abce3-0abf-4070-a8ca-e9aec8aaa004
📒 Files selected for processing (8)
apps/harness/lib/harness/providers/cursor_session.exapps/harness/lib/harness/providers/opencode_session.exapps/harness/lib/harness/session_manager.exapps/harness/test/harness/storage_test.exsapps/server/src/orchestration/Layers/ProviderCommandReactor.tspackages/shared/src/DrainableWorker.test.tspackages/shared/src/DrainableWorker.tsscripts/stress-test-resume-multi.ts
…inistic wait - Reject all pending promises on schema error (id="unknown"), not just oldest - Drop ephemeral port from OpenCode persist_binding (stale after restart) - Replace sleep(5000) with deterministic wait polling sessionStopped tracker Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
processSessionStopRequested calls providerService.stopSession() which for Claude interrupts the stream fiber via Fiber.interrupt(). This interruption propagated through processDomainEventSafely (which re-throws interrupt-only causes) and killed the DrainableWorker and the entire Node server. Fix: wrap stopSession in Effect.uninterruptible + Effect.ignore so the stop completes without propagating the fiber interrupt upward. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/server/src/orchestration/Layers/ProviderCommandReactor.ts`:
- Around line 709-713: The stopSession call currently uses Effect.ignore which
hides real failures; replace the ignore with logic that captures the failure
cause from providerService.stopSession(threadId) and uses
Cause.hasInterruptsOnly to distinguish interrupt-only failures from real
failures: if Cause.hasInterruptsOnly(cause) then swallow/ignore, otherwise call
appendProviderFailureActivity with "provider.session.stop.failed" (include the
cause/details) and then rethrow or fail the effect so the non-interrupt failure
is not silently dropped and the subsequent unconditional marking of the thread
as "stopped" does not occur. Use providerService.stopSession,
Cause.hasInterruptsOnly, and appendProviderFailureActivity to implement this
behavior.
- Around line 780-784: The current worker created by
makeDrainableWorker(processDomainEventSafely, { concurrency: 8 }) allows events
for the same threadId to run concurrently and can race lifecycle operations;
modify processing to serialize per-threadId by sharding or mutexing: introduce a
per-thread queue/lock keyed by threadId (used by ProviderSessionDirectory and
ProviderService operations) and ensure processDomainEventSafely first acquires
the threadId shard or mutex before handling events and releases it after, or
replace the flat worker with a set of N shard workers where events are
dispatched to shard = hash(threadId) so each shard runs with single concurrency;
update references to makeDrainableWorker, processDomainEventSafely,
ProviderSessionDirectory, ProviderService, and threadId accordingly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 88fab8ec-9ddb-4128-bb8c-2e320fe3f4bd
📒 Files selected for processing (1)
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
Comment on lines
+709
to
+713
| // Run stop as uninterruptible to prevent Claude SDK fiber interruptions | ||
| // from propagating up and killing the reactor worker. | ||
| yield* providerService | ||
| .stopSession({ threadId: thread.id }) | ||
| .pipe(Effect.uninterruptible, Effect.ignore); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Verify current stop handling in reactor
rg -n -C4 'stopSession\(\{ threadId: thread\.id \}\)|Effect\.uninterruptible|Effect\.ignore' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Verify stopSession side effects that should not be silently dropped
rg -n -C6 'const stopSession|adapter\.stopSession|directory\.remove|analytics\.record' \
apps/server/src/provider/Layers/ProviderService.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 6303
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check if appendProviderFailureActivity exists and where it's used
rg -n 'appendProviderFailureActivity|hasInterruptsOnly' \
apps/server/src/orchestration/ apps/server/src/provider/
# Check broader error handling in reactor for similar patterns
rg -n 'Effect\.catch|Effect\.tapError|Effect\.handle' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Verify thread session state setting
rg -n -A2 'setThreadSession' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -20Repository: Ranvier-Technologies/t3code-OTP
Length of output: 1953
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Examine appendProviderFailureActivity definition and signature
rg -n -A15 'const appendProviderFailureActivity = ' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check one of the existing error handling patterns using appendProviderFailureActivity + Effect.catchCause
rg -n -B2 -A8 'yield\* appendProviderFailureActivity\({' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -40
# Verify what happens after stopSession in the reactor
sed -n '708,725p' apps/server/src/orchestration/Layers/ProviderCommandReactor.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 848
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check existing error handling patterns with appendProviderFailureActivity
rg -n -B2 -A8 'appendProviderFailureActivity\({' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -60
# Verify what happens after stopSession in the reactor
sed -n '708,725p' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check if there are similar stop patterns elsewhere in the codebase
rg -n -B3 -A3 'Effect\.catchCause.*hasInterruptsOnly' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 202
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check existing error handling patterns with appendProviderFailureActivity
rg -n -B2 -A8 'appendProviderFailureActivity' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -80
# Verify what happens after stopSession in the reactor
sed -n '708,725p' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check if there are similar stop patterns elsewhere in the codebase
rg -n -B3 -A5 'hasInterruptsOnly' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 4711
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Compare the error handling pattern at line 770 with what's proposed
sed -n '767,780p' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check if Cause is imported from Effect
rg -n '^import.*Cause|from.*effect' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -15
# Verify the stopSession function definition for exact side effects
sed -n '442,461p' apps/server/src/provider/Layers/ProviderService.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 1443
Don't ignore non-interrupt stopSession failures.
Effect.ignore hides real stop/cleanup failures (adapter stop, directory cleanup, analytics), but the thread session is unconditionally marked "stopped" afterward (line 716), risking provider/orchestration state desync.
The codebase already uses Cause.hasInterruptsOnly (line 770) to distinguish interrupt-only failures from real failures, and appendProviderFailureActivity already supports "provider.session.stop.failed" (line 175). Apply the same pattern here.
Proposed fix
- yield* providerService
- .stopSession({ threadId: thread.id })
- .pipe(Effect.uninterruptible, Effect.ignore);
+ yield* providerService
+ .stopSession({ threadId: thread.id })
+ .pipe(
+ Effect.uninterruptible,
+ Effect.catchCause((cause) =>
+ Cause.hasInterruptsOnly(cause)
+ ? Effect.void
+ : appendProviderFailureActivity({
+ threadId: thread.id,
+ kind: "provider.session.stop.failed",
+ summary: "Provider session stop failed",
+ detail: Cause.pretty(cause),
+ turnId: null,
+ createdAt: now,
+ }).pipe(Effect.zipRight(Effect.failCause(cause))),
+ ),
+ );🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/server/src/orchestration/Layers/ProviderCommandReactor.ts` around lines
709 - 713, The stopSession call currently uses Effect.ignore which hides real
failures; replace the ignore with logic that captures the failure cause from
providerService.stopSession(threadId) and uses Cause.hasInterruptsOnly to
distinguish interrupt-only failures from real failures: if
Cause.hasInterruptsOnly(cause) then swallow/ignore, otherwise call
appendProviderFailureActivity with "provider.session.stop.failed" (include the
cause/details) and then rethrow or fail the effect so the non-interrupt failure
is not silently dropped and the subsequent unconditional marking of the thread
as "stopped" does not occur. Use providerService.stopSession,
Cause.hasInterruptsOnly, and appendProviderFailureActivity to implement this
behavior.
Comment on lines
+780
to
+784
| // Allow up to 8 concurrent event processors so independent thread operations | ||
| // (session starts, turn dispatches) don't block each other. Events for the | ||
| // same thread are still safe: ensureSessionForThread and sendTurnForThread | ||
| // operate on per-thread state and the provider layer serializes per-session. | ||
| const worker = yield* makeDrainableWorker(processDomainEventSafely, { concurrency: 8 }); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect DrainableWorker ordering/concurrency semantics
rg -n -C5 'makeDrainableWorker|concurrency|enqueue|dequeue|Queue|take|drain|fork' \
packages/shared/src/DrainableWorker.ts
# Inspect tests for any per-key/per-thread ordering guarantees
rg -n -C5 'concurrency|order|fifo|thread|enqueue' \
packages/shared/src/DrainableWorker.test.ts
# Confirm all handled events here are thread-scoped (for safe sharding by threadId)
rg -n -C3 'type ProviderIntentEvent|threadId|thread\.' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 25720
🏁 Script executed:
# Check orchestrationEngine.dispatch for per-thread serialization
rg -n -C8 'dispatch.*function|dispatch.*Effect|orchestrationEngine\.dispatch' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts \
| head -60
# Search for Mutex, Ref, or other synchronization per threadId
rg -n 'Mutex|Ref\.make.*threadId|Lock|Semaphore|serialization' \
apps/server/src/orchestration/
# Check providerManager.ts for per-thread coordination
fd -t f providerManager.ts apps/server/src/ 2>/dev/null || echo "Searching..."
rg -l 'providerManager|ProviderManager' apps/server/src/orchestration/
# Look for how events are enqueued (single queue vs sharded)
rg -n -C5 'worker\.enqueue|selectWorker|enqueue.*event' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.tsRepository: Ranvier-Technologies/t3code-OTP
Length of output: 2244
🏁 Script executed:
# Find orchestrationEngine implementation
fd -t f -e ts 'orchestrationEngine|Orchestration.*Engine' apps/server/src/ | head -20
# Directly check for files that define OrchestrationEngine
rg -l 'export.*OrchestrationEngine|class OrchestrationEngine|interface OrchestrationEngine' apps/server/src/
# Look at how ProviderCommandReactor is initialized and what orchestrationEngine is
rg -n -B10 'const.*orchestrationEngine' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -30
# Search for providerManager in server src
fd -t f providerManager apps/server/src/Repository: Ranvier-Technologies/t3code-OTP
Length of output: 1471
🏁 Script executed:
# Read OrchestrationEngine.ts Services implementation to check dispatch
cat -n apps/server/src/orchestration/Services/OrchestrationEngine.ts | head -200
# Search for dispatch definition and any per-threadId synchronization
rg -n -A15 'dispatch.*Effect|dispatch.*function' \
apps/server/src/orchestration/Services/OrchestrationEngine.ts | head -80
# Search for Mutex, Ref, or Map patterns around threadId in OrchestrationEngine
rg -n 'Mutex|Ref\.make|Map.*threadId|per-thread|serialize' \
apps/server/src/orchestration/Services/OrchestrationEngine.ts
# Find providerManager or providerService
fd -t f 'provider.*[Mm]anager|[Pp]rovider.*[Ss]ervice' apps/server/src/ | grep -E 'Service|Manager'Repository: Ranvier-Technologies/t3code-OTP
Length of output: 3549
🏁 Script executed:
# Find ProviderService or providerManager
rg -l 'providerService|ProviderService' apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check the ProviderService implementation
cat -n apps/server/src/provider/Services/ProviderService.ts | head -150
# Search for Mutex/Ref/serialization around threadId in ProviderCommandReactor
rg -n -B5 -A5 'Mutex|Ref\.make|Map.*threadId' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
# Check if ensureSessionForThread is wrapped in Mutex or per-thread locking
rg -n -B3 -A20 'const ensureSessionForThread' \
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts | head -50Repository: Ranvier-Technologies/t3code-OTP
Length of output: 5243
🏁 Script executed:
# Search for any reference to providerManager in codebase
rg -l 'providerManager|ProviderManager' apps/server/src/
# List all files in provider directory
find apps/server/src/provider -type f -name '*.ts' | head -20
# Check if there's a manager or coordinator anywhere
fd -t f 'manager|Manager|coordinator|Coordinator' apps/server/src/provider/ apps/server/src/orchestration/
# Check the guideline file path and content again
cat -n coding_guidelines || echo "No coding_guidelines file at root"Repository: Ranvier-Technologies/t3code-OTP
Length of output: 1315
🏁 Script executed:
# Read ProviderSessionDirectory to check per-session/thread locking
cat -n apps/server/src/provider/Layers/ProviderSessionDirectory.ts | head -200
# Search for Mutex or per-thread locking in ProviderSessionDirectory
rg -n 'Mutex|Ref\.make|Queue|serialize|lock|Lock' \
apps/server/src/provider/Layers/ProviderSessionDirectory.ts | head -30
# Read ProviderService Layer implementation to see dispatch coordination
rg -n -A30 'export const.*make|Effect.gen.*function' \
apps/server/src/provider/Layers/ProviderService.ts | head -100Repository: Ranvier-Technologies/t3code-OTP
Length of output: 11508
Concurrency: 8 needs per-thread ordering guarantees to prevent lifecycle event races.
Same-thread events can execute concurrently and cause race conditions (e.g., stop vs turn-start). The ProviderSessionDirectory and ProviderService lack per-threadId serialization. Recommend sharding by threadId (single-concurrency per shard) or wrapping handlers in per-thread Mutex.
Safer pattern (sharded workers, per-thread serialization)
+ const EVENT_WORKER_COUNT = 8;
+ const workers = yield* Effect.all(
+ Array.from({ length: EVENT_WORKER_COUNT }, () =>
+ makeDrainableWorker(processDomainEventSafely, { concurrency: 1 }),
+ ),
+ );
+ const hashThreadId = (threadId: ThreadId): number =>
+ [...threadId].reduce((acc, ch) => ((acc * 31 + ch.charCodeAt(0)) | 0), 0);
+ const selectWorker = (threadId: ThreadId) =>
+ workers[Math.abs(hashThreadId(threadId)) % EVENT_WORKER_COUNT]!;
- const worker = yield* makeDrainableWorker(processDomainEventSafely, { concurrency: 8 });- return worker.enqueue(event);
+ return selectWorker(event.payload.threadId).enqueue(event);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/server/src/orchestration/Layers/ProviderCommandReactor.ts` around lines
780 - 784, The current worker created by
makeDrainableWorker(processDomainEventSafely, { concurrency: 8 }) allows events
for the same threadId to run concurrently and can race lifecycle operations;
modify processing to serialize per-threadId by sharding or mutexing: introduce a
per-thread queue/lock keyed by threadId (used by ProviderSessionDirectory and
ProviderService operations) and ensure processDomainEventSafely first acquires
the threadId shard or mutex before handling events and releases it after, or
replace the flat worker with a set of N shard workers where events are
dispatched to shard = hash(threadId) so each shard runs with single concurrency;
update references to makeDrainableWorker, processDomainEventSafely,
ProviderSessionDirectory, ProviderService, and threadId accordingly.
ClaudeAdapter.stopSessionInternal used yield* Fiber.interrupt(streamFiber) which joins the dying fiber and propagates the interruption Exit up through the fiber tree, crashing the entire Node server. Changed to forkChild so the interrupt fires without joining. Also added --exclude <provider> flag to stress test script for isolating provider-specific issues during debugging. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Codex CLI sends exec_approval_request even with approvalPolicy: "never" on resumed sessions (especially when the model changes, e.g. gpt-5.3-codex → gpt-5.4). The harness now auto-approves all non-user-input RPC requests in full-access runtime mode, matching OpenCode's auto-approve pattern. This was the root cause of the Codex resume test failure — the turn hung waiting for an approval response that never came because the event was unmapped in the Node layer. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Codex CLI's CommandExecutionRequestApprovalResponse serde deserializer expects the variant "approve" (without trailing 'd'). Our auto-approve was sending "approved" which caused: "failed to deserialize: unknown variant 'approved'" — the approval was rejected by the sandbox. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
🧹 Nitpick comments (1)
apps/harness/lib/harness/providers/codex_session.ex (1)
843-846: Emitting "request/resolved" without a prior pending entry creates an orphan resolve event.When auto-approving, the code emits
"request/resolved"without ever adding an entry tostate.pendingor emitting a "request opened" event. While downstream handlers safely handle this (Map.delete on non-existent key is a no-op, DELETE with zero rows succeeds), this creates a semantic inconsistency:
- The projector receives a resolve for a request that was never tracked
- The TypeScript adapter (HarnessClientAdapter) emits a
request.resolvedruntime event without a correspondingrequest.openedeventThis is likely intentional (auto-approved requests shouldn't appear pending to users), but worth documenting in a brief comment for future maintainers, or omitting the resolve event entirely since no request was ever surfaced.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/harness/lib/harness/providers/codex_session.ex` around lines 843 - 846, The code emits a "request/resolved" event via emit_event(state, :notification, "request/resolved", ...) for auto-approved requests even though no entry is added to state.pending or a "request opened" event is ever emitted; update the code around emit_event (and the auto-approve branch) to either remove the resolve emission entirely or add a concise comment explaining this is intentional (auto-approved requests are not tracked in state.pending and therefore have no corresponding "request.opened" event), referencing emit_event, "request/resolved", and state.pending so future maintainers understand the semantic choice.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@apps/harness/lib/harness/providers/codex_session.ex`:
- Around line 843-846: The code emits a "request/resolved" event via
emit_event(state, :notification, "request/resolved", ...) for auto-approved
requests even though no entry is added to state.pending or a "request opened"
event is ever emitted; update the code around emit_event (and the auto-approve
branch) to either remove the resolve emission entirely or add a concise comment
explaining this is intentional (auto-approved requests are not tracked in
state.pending and therefore have no corresponding "request.opened" event),
referencing emit_event, "request/resolved", and state.pending so future
maintainers understand the semantic choice.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ed341303-63ac-4f57-8288-29f9c90853c4
📒 Files selected for processing (1)
apps/harness/lib/harness/providers/codex_session.ex
The Codex app-server protocol defines CommandExecutionRequestApprovalResponse as a Rust serde enum with variants: accept, acceptForSession, decline, cancel. Neither "approve" nor "approved" are valid — the deserializer rejects both. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
oxfmt --check flagged docs/index.html, docs/pitch.html, and scripts/stress-test-resume-multi.ts. Also added a comment explaining why the auto-approve branch emits request/resolved without a prior request.opened event — intentional since auto-approved requests bypass state.pending entirely. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
3 tasks
This was referenced Mar 26, 2026
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.
Summary
P2 — Cursor/OpenCode bindings: Added
persist_bindingtocursor_session.exandopencode_session.exso resume works for all harness providers (not just Codex). Fixednormalize_resume_cursorto re-encode Cursor/OpenCode cursors as JSON strings — their session modules callJason.decodeonresumeCursor, so passing a decoded map caused silent failures.P0 — Reactor concurrency: Added optional
concurrencyparameter toDrainableWorker(default 1). Set to 8 inProviderCommandReactorso independent thread operations (session starts, turn dispatches) run in parallel. Fixes the 12-session resume Phase 3 timeout where sequential session starts exceeded the 120s deadline.P1 — Thread visibility: Stress test now reuses
bootstrapProjectIdfrom the welcome payload instead of creating a separate project. Threads persist under the main sidebar project across page refreshes.Test plan
mix test test/harness/storage_test.exs— 38/38 (binding normalization)bun vitest run packages/shared/src/DrainableWorker.test.ts— 2/2 (serial + concurrent)bun vitest run apps/server/src/orchestration— 103/103bun turbo run typecheck --filter=@t3tools/shared --filter=t3— 4/4 cleanbun run scripts/stress-test-resume-multi.ts multi— validate 4-provider resume E2E🤖 Generated with Claude Code
Summary by CodeRabbit
Bug Fixes
New Features
Performance Improvements
Tests