fix(terminal): queue concurrent send writers instead of fail-fast spin

[Raghav-Bansal-15]

Summary

• TerminalSession.send's tryLock-fail storm is replaced by a fair ReentrantLock with a bounded 5s wait (injectable via inputQueueTimeoutMillis)
• writers that outwait the bound are shed with RESOURCE_EXHAUSTED carrying a google.rpc RetryInfo retry-delay instead of an instant spin-inducing failure; interruption surfaces as CANCELLED

Test plan

• TerminalSendQueueTest: concurrent writers park for the bound then shed with retry-after; queued writer proceeds once the pipe drains
• :boss-app-terminal:test (17 tests), detekt, ktlint all pass

[fluck-boss]

@claude please review concurrent terminal writers, focusing on queue fairness, cancellation, backpressure, shutdown, and preservation of write ordering.

[fluck-boss]

This branch now conflicts with dev in modules/boss-app-terminal/.../TerminalSession.kt and needs a rebase I can't push for you (my token can't carry dev's workflow-file updates). git fetch origin && git rebase origin/dev, then force-push. The two conflicts and their resolutions:

1. Imports: dev added java.util.concurrent.atomic.AtomicBoolean, you added java.util.concurrent.TimeUnit - keep BOTH.
2. Constructor: dev added inputWriteTimeoutMillis: Long = 5_000, your branch adds inputQueueTimeoutMillis: Long = INPUT_QUEUE_TIMEOUT_MILLIS - these are different knobs (write stall vs queue wait), so keep BOTH parameters; the queue writer you add should use yours, the existing write path keeps dev's.

CI was green on the current head before the conflict appeared, so a clean rebase should go straight back to review.

[fluck-boss]

@claude please review this exact head (6782f29) for merge into dev. Check that the change is correct and complete for its stated goal, that tests cover the risk being fixed, and flag any security issues (secrets, injection, auth or approval bypass, path handling). Keep the review focused on this diff.

[claude]

Claude finished @fluck-boss's task in 3m 35s —— View job

---

Review of 6782f29 — fix(terminal): queue concurrent send writers instead of fail-fast spin

• Gather context (read diff, TerminalSession.kt, TerminalServiceImpl.kt, module build config)
• Review correctness/completeness vs stated goal
• Assess test coverage of the risk
• Security pass (secrets, injection, auth/approval bypass, path handling)
• Post review

---

The core change is right and the direction is the correct one: tryLock() → tryLock(bound) on a fair lock, with the timeout knob injectable, is exactly how you turn a fail-fast spin into shedding. test 2 is a genuine regression test — against the old code queued.get(200ms) completes exceptionally rather than timing out, so it fails.

What I'd want resolved before this goes into dev is that the caller is withContext(Dispatchers.IO), and that changes what "bounded 5s wait" actually costs. Details below, most-important first.

---

1. Cancellation does not reach the parked writer, so the CANCELLED path is effectively dead code

TerminalServiceImpl.kt:129-136:

override suspend fun sendInput(request: SendInputRequest): Empty =
    withContext(Dispatchers.IO) {
        ...
        session(request.sessionId).send(request.data.toByteArray())

Dispatchers.IO does not interrupt its threads on coroutine cancellation — only runInterruptible does. So when the gRPC client cancels, hits its deadline, or drops the connection, the thread stays parked in inputLock.tryLock(5s) for the full bound. Two consequences:

• The InterruptedException → CANCELLED branch (TerminalSession.kt:115-119) is unreachable in production. It is also the one behaviour named in the PR description that has no test.
• A cancelled send still acquires the lock and writes the bytes, up to 5 s after the caller gave up. Previously fail-fast bounded that window to ~0. The bytes were authorized when sent, so this isn't an approval bypass, but "input the operator cancelled still lands in the shell" is a new window worth being deliberate about.

createSession already reaches for currentCoroutineContext().ensureActive() (TerminalServiceImpl.kt:49), so the pattern exists in this file. The stronger fix is runInterruptible(Dispatchers.IO) { session(...).send(...) }, which makes the InterruptedException handler real and lets a cancelled RPC release its thread immediately. Fix this →

2. The waiter set is bounded in time but not in count — this can starve Dispatchers.IO process-wide

Dispatchers.IO is capped at 64 threads. Every queued writer now pins one for up to 5 s, where before lock contention pinned none. 64 concurrent sendInput calls against one busy session therefore stall every other IO-dispatched operation in the terminal service process, createSession included. That is an availability amplification introduced by this change, and it's the part of the reviewer's "backpressure" question that is still open.

This module already has the idiom for it: activeSlots = Semaphore(16) and streamSlots = Semaphore(32) bound concurrency by count and shed instantly past the cap. A Semaphore(N) around the queue wait — shed with the same RESOURCE_EXHAUSTED + RetryInfo immediately when there are already N waiters — would make the queue bounded in both dimensions and keep the shed path cheap under real flood. Fix this →

3. Fairness is not write ordering, and the comment claims it is

TerminalSession.kt:34 — "Fair so queued writers drain in arrival order instead of barging past one another."

ReentrantLock(true) is FIFO over lock-arrival order, and lock-arrival order here is whatever order the IO-pool threads happen to reach tryLock. sendInput is a unary RPC dispatched concurrently, so two keystrokes sent back to back can reach the lock inverted, and the fair lock will then faithfully preserve the wrong order. Fairness buys anti-starvation, not ordering — which is genuinely worth having, since the previous barging tryLock() could starve one writer indefinitely. Please reword so nobody later builds on an ordering guarantee that isn't there. (The hazard is pre-existing; only the claim is new.)

4. No jitter on the retry hint

Every shed writer gets an identical retryDelay of the full queue timeout, so all N wake together and re-form the same convoy at t+5s. That converts a tight spin into a synchronized lockstep retry rather than removing the convoy. A small random jitter on the delay is the usual answer.

Relatedly, the delay is the whole bound the caller just finished waiting out — defensible as a heuristic, but it means a client honouring RetryInfo sees ~10 s end-to-end for one keystroke.

5. A shed writer on a dying session is told to retry

The !active || !process.isAlive check (:103) happens only after acquiring. So during close() / terminate(), writers parked behind a stuck write() wait the full bound and then get RESOURCE_EXHAUSTED "Terminal input is busy" with a retry-after for a terminal that has exited, instead of FAILED_PRECONDITION. Also on the shutdown path specifically: TerminalServiceImpl.close() doesn't wake queued writers, and its 5 s process-exit deadline is now raced by a 5 s input queue, so IO threads stay occupied for up to 5 s after close() returns. Neither is a correctness bug, but shutdown was an explicit review focus and both are cheap to improve (re-check active before parking, and/or have terminate() flip a flag the queue observes).

6. Interaction with dev's inputWriteTimeoutMillis (relevant to the rebase you were asked for)

dev added a write-stall bound of 5 s; this adds a queue bound of 5 s. Keeping both — which is the right resolution — makes worst-case occupancy of one IO thread by one sendInput 10 s, not 5. Worth stating in the comment or in the constructor KDoc, since the two knobs compose and only their sum is the real bound. It also means a waiter behind two stalled writers can never succeed, which is fine (that's shedding working), but the default values should be chosen with the sum in mind.

7. Smaller things

• import com.google.protobuf.Any (:7) shadows kotlin.Any for the whole file. The file already dodges exactly this hazard for Status by fully qualifying com.google.rpc.Status at :124. Do the same for Any (or import com.google.protobuf.Any as ProtoAny) — TerminalServiceImpl.kt:27 is private val lock = Any(), so this is a live footgun for anyone moving code between the two files.
• No validation on inputQueueTimeoutMillis. A zero or negative value silently restores non-blocking tryLock, and a negative one builds a proto Duration with negative nanos, which violates the Durations invariant that seconds and nanos share a sign. It's a private, default-valued param on an internal class so the blast radius is small, but require(inputQueueTimeoutMillis > 0) is one line.
• Undeclared transitive dependency. com.google.rpc.RetryInfo comes from proto-google-common-protos and io.grpc.protobuf.StatusProto from grpc-protobuf; both reach this module only via api(libs.grpc.protobuf) in :boss-ipc. It compiles, but this module is also built by nativeCompile with --no-fallback — worth confirming the native image resolves the RetryInfo descriptor for Any.pack, since the JVM test won't tell you.

Tests

The two tests are real and test 2 has a clean mutation story. Gaps:

• Fairness is unpinned. Nothing fails if ReentrantLock(true) becomes ReentrantLock(). That argument is the anti-starvation property; a test that a queued writer eventually proceeds while others keep arriving would pin it.
• CANCELLED on interrupt is untested — and per finding 1 it may not be reachable at all from the real caller. It's the one claim in the PR description with no coverage.
• No test that a writer acquiring the lock after exit gets FAILED_PRECONDITION, which is what distinguishes it from the new shed path.
• CompletableFuture.runAsync on blocking work is fragile here. These tasks block without a ManagedBlocker, so the common ForkJoinPool creates no compensation threads. CompletableFuture falls back to thread-per-task only when getCommonPoolParallelism() <= 1; on a 3-core runner parallelism is 2, so test 1's permanently-blocked holder pins one worker and the 6 "concurrent" writers then run one at a time on the remaining worker. The assertions still pass (~1.8 s, well inside the 30 s latch), but the test stops exercising the concurrency in its own name, and test 2 (holder + queued = exactly 2) is sitting right on that edge. An explicit Executors.newFixedThreadPool(writers + 1), shut down in finally, makes both independent of runner core count. Fix this →

I did not run the suite locally (no network for the Gradle deps in this checkout); CI's :boss-app-terminal:test result stands.

Security

Nothing blocking.

• Secrets / logging — none. inputBusy() carries no session id, no payload; RetryInfo leaks nothing.
• Injection / path handling — untouched by this diff. validateLaunchInput, the NUL and = checks, and IpcEnvironment.removeCredentials are all unchanged.
• Auth / approval bypass — none. session(id) performs IpcCall.current() and IpcCall.requireOwner(found.ownerInstance) before send is reached, so the queue wait happens strictly after authorization and an unauthorized caller never parks on the lock.
• The only security-relevant deltas are availability (finding 2 — a caller can now pin IO threads by flooding one session, where fail-fast made that impossible) and the cancelled-write window (finding 1).

---

Recommendation: findings 1 and 2 before merge — they're the two that change the process's behaviour under exactly the load this PR is about. 3 (reword) and 7 are cheap. 4, 5, 6 and the test gaps are worth a follow-up if you'd rather land the shedding first. Note the branch still needs the rebase @fluck-boss described; finding 6 is about how the two timeouts compose once both are present.
• fix/b23-terminal-input-queue

[fluck-boss]

Thanks for the fix. Two things before this can land:

1. This branch now conflicts with dev (the terminal input path moved underneath it). Please rebase onto current dev and resolve.
2. The automated review raised a real correctness concern with the queue: parked send writers can pin Dispatchers.IO threads under load, and a client cancellation may not interrupt a parked writer, so a later write can still go through after the caller has given up. Please address both - e.g. bound the parked-writer count (or offload parking off Dispatchers.IO) and make cancellation abandon the queued write.

Converting to draft in the meantime - flip it back to ready when CI is green on the rebased branch.

[Raghav-Bansal-15]

Both review points are addressed on the rebased branch (head 60415ef86, now on current dev).

Rebase. Resolved the TerminalSession.kt conflicts as flagged: imports keep both AtomicBoolean (dev's bounded write path) and the queue additions, and the constructor keeps both inputWriteTimeoutMillis (write stall bound, unchanged) and inputQueueTimeoutMillis (queue wait bound). dev's inputClosed/requireUsableInput/discardInput write path is preserved as-is.

Parked writers. The fair ReentrantLock.tryLock(5s) is replaced by a kotlinx.coroutines.Mutex with a bounded withTimeoutOrNull acquisition, and send is now suspend:

• Queued writers suspend instead of parking a Dispatchers.IO thread, so a stalled pipe cannot pin the pool under load — there is no parked-thread count left to bound.
• Coroutine cancellation is delivered at the lock() suspension point, so a cancelled call abandons its queued write instead of sending it after the caller gave up; it surfaces as CANCELLED through the normal gRPC path. An ensureActive() check inside the critical section covers a cancellation racing the grant, before writeInput runs.
• A grant that races the timeout shed is handed back via Mutex.holdsLock(ticket)/unlock(ticket) so the queue can't wedge behind a dead caller.
• The shed path is unchanged: RESOURCE_EXHAUSTED with google.rpc.RetryInfo after inputQueueTimeoutMillis, and FIFO drain order is preserved by the fair mutex.

Tests. TerminalSendQueueTest keeps the shed-with-retry-after and drain-on-release cases (now also asserting arrival-order writes) and adds two: a queued writer on a single-thread dispatcher leaves the thread free for a probe coroutine (no pinning), and cancelAndJoin on a queued writer leaves its bytes unsent after the pipe drains. TerminalPipeFailureTest was updated for the suspend signature.

./gradlew :boss-app-terminal:test :boss-app-terminal:detekt :boss-app-terminal:ktlintCheck is green locally. Will flip back to ready once CI is green.

[fluck-boss]

Pushed 723df2b on top of 4ca5991. The rebase had dropped '\u0000' !in name from validateLaunchInput (it came in with #776 and isn't related to the queue change), so this puts that one check back. Nothing else changed.

@claude please review this exact head (723df2b) for merge into dev. Confirm your earlier points on 6782f29 are resolved by the Mutex rewrite (cancellation reaching queued writers, no parked IO threads, the grant-after-timeout race, write ordering), check that the tests cover them, and flag any security issues. Keep it focused on this diff.

[claude]

Claude finished @fluck-boss's task in 4m 50s —— View job

---

Review of 723df2b — Mutex rewrite of the terminal send queue

• Gather context (diff, TerminalSession.kt, TerminalServiceImpl.kt, BossIpcServer.kt, module build)
• Verify earlier findings resolved (cancellation, IO threads, grant-after-timeout race, ordering)
• Assess test coverage
• Security pass
• Post review

---

The Mutex rewrite is the right call and it resolves both blocking findings. suspend fun send + Mutex.lock(ticket) under withTimeoutOrNull is a strictly better shape than the fair ReentrantLock.tryLock(bound) it replaces, and the two new tests are real regression tests for exactly the two things review asked for. The NUL-check restoration in 723df2b is a clean revert to dev's validateLaunchInput — git diff of the whole branch against its merge base shows that line unchanged, which is the right shape for "put it back".

Recommendation: mergeable. Nothing below blocks; #1 and #2 are one-liners I'd take in this PR, the rest are follow-up.

---

Earlier findings, status

#	Finding on 6782f29	Status on 723df2b
1	Cancellation never reaches a parked writer; cancelled send still writes	✅ Resolved
2	Waiters pin Dispatchers.IO threads (cap 64)	✅ Resolved for threads — residual is memory, see #5
3	Grant-after-timeout race	✅ Handled — see note below
4	Fairness ≠ write ordering, comment claimed it	⚠️ class comment fixed; claim survives in the PR description and a test comment
5–9	jitter, dying-session status, composed bound, Any shadowing, knob validation, undeclared dep	❌ unchanged

Cancellation (#1). inputMutex.lock(ticket) is a suspension point, so a cancelled RPC now abandons its queued write rather than landing bytes up to 5 s after the caller gave up. currentCoroutineContext().ensureActive() at TerminalSession.kt:114 closes the grant-races-cancellation window before writeInput runs. The residual window — cancellation arriving after ensureActive() and during the blocking writeInput — is inherent and much narrower than what it replaces. TerminalPipeFailureTest confirms the suspend conversion didn't disturb the ABORTED/FAILED_PRECONDITION paths.

IO threads (#2). Waiters suspend, so withContext(Dispatchers.IO) in sendInput releases its thread. Worst-case IO occupancy is now one thread per holder, and activeSlots = Semaphore(16) bounds sessions, so ≤16 threads rather than 64. That's the fix.

Grant-after-timeout (#3). Correct, and the fresh kotlin.Any() ticket per call is load-bearing in a way worth not losing: Mutex.lock(owner) throws IllegalStateException if that same owner already holds it, so a shared owner object would have been a bug here. Two notes on the defensive unlock at :128:

• It is unreachable on real cancellation. withTimeoutOrNull only swallows its own TimeoutCancellationException; an externally cancelled job rethrows straight out of acquireInputLock, skipping the holdsLock check. So the comment's "a dead caller" case is precisely the one this line does not cover.
• It doesn't need to. kotlinx 1.11's Mutex releases a permit granted concurrently with cancellation (onCancellationRelease), on both the timeout and the cancellation path. So this is belt-and-braces on a guarantee the library already makes — fine to keep, but reword the comment so nobody later concludes the invariant rests on this line.

---

1. A non-positive inputQueueTimeoutMillis bricks input entirely (was finding 7b, now worse)

withTimeoutOrNull returns null without running the block when timeMillis <= 0. So under the old tryLock a zero knob degraded to non-blocking acquisition; here it never attempts the lock at all, and every sendInput fails RESOURCE_EXHAUSTED on a completely idle session. One line in init:

require(inputQueueTimeoutMillis > 0)

(Correcting myself from the last review: I claimed a negative value would build a Duration violating the shared-sign invariant. That was wrong — Java's truncating division and remainder keep seconds and nanos aligned in sign, so RetryInfo stays valid. The denial-of-input above is the real consequence.) Fix this →

2. import com.google.protobuf.Any is now actively costing you (was 7a)

TerminalSession.kt:7 shadows kotlin.Any for the whole file, and the rewrite made that visible three times over: kotlin.Any() at :109 and :122, and the fully-qualified com.google.rpc.Status at :147. import com.google.protobuf.Any as ProtoAny removes all three qualifications and the footgun — TerminalServiceImpl.kt:27 is private val lock = Any(), so code moved between these two files changes meaning silently. Fix this →

3. send is a suspend function that still blocks, with no dispatcher of its own

writeInput does writer.join(inputWriteTimeoutMillis) — up to 5 s of blocking on whatever thread the caller happens to be on. In production that's fine because TerminalServiceImpl.sendInput wraps in withContext(Dispatchers.IO), but the contract now lives in the caller rather than in send. Worth noting because test 3 launches a writer on a single-thread dispatcher and only passes because that writer never acquires the lock — if it did, it would block that thread for the write bound. A withContext(Dispatchers.IO) (or runInterruptible) around writeInput inside send would make the blocking local to where it happens.

4. The queue bound equals the write bound, so a stalled write sheds every waiter by construction

Both default to 5 s. A holder that stalls aborts at t=5 s and every waiter sheds at t=5 s — the waiters can never benefit from the holder's own abort-and-close path, which is the one case where waiting would have paid off. Worst-case occupancy of one sendInput is their sum, 10 s, which is still undocumented on either constructor param. Either default the queue bound above the write bound, or say in the KDoc that the two compose.

5. Waiter count is unbounded, and each waiter retains its payload

The residual of finding 2, now in memory rather than threads. TerminalServiceImpl.kt:146 calls request.data.toByteArray() before send, so each suspended waiter holds a copy of up to 64 KiB (plus the SendInputRequest itself) across the full bound. BossIpcServer doesn't set maxConcurrentCallsPerConnection, so grpc-netty's default leaves the stream count effectively unbounded.

Severity is much lower than the thread version: session(id) runs IpcCall.current() and IpcCall.requireOwner(found.ownerInstance) before send, so this needs an already-authenticated session owner — a buggy or runaway client, not an attacker. A Semaphore(N) on waiters in the activeSlots/streamSlots idiom, shedding instantly past the cap, would close it; a follow-up is fine.

6. The ordering claim survives outside the code

The class comment at :39-41 correctly dropped it, but the PR description still says "FIFO drain order is preserved by the fair mutex" and TerminalSendQueueTest.kt:99 comments "Queued writers drain in arrival order". kotlinx's Mutex is FIFO over lock-arrival order, and since sendInput is a unary RPC dispatched concurrently onto the IO pool, two keystrokes sent back to back can reach lock() inverted — the mutex will then faithfully preserve the wrong order. The property you have is anti-starvation, which is genuinely worth having. Hazard is pre-existing; only the claim is new.

7. Still unchanged from the last review

• No jitter on retryDelay — every shed writer gets the identical full bound, so all N wake together at t+5s and re-form the convoy.
• A shed writer on a dying session is told to retry. requireUsableInput() runs only after acquisition, so during close()/terminate() waiters get RESOURCE_EXHAUSTED + retry-after for a terminal that has exited, instead of FAILED_PRECONDITION. Cheaper now that they're suspended rather than pinning threads, but still the wrong answer.
• Undeclared transitive dependency. com.google.rpc.RetryInfo and io.grpc.protobuf.StatusProto reach this module only via api(libs.grpc.protobuf) in :boss-ipc; boss-app-terminal/build.gradle.kts declares neither. It compiles, and Any.pack needs no reflection so nativeCompile --no-fallback should be fine, but declaring them is honest.

---

Tests

Good coverage of the two things that mattered:

• Test 3 (suspends without pinning its dispatcher thread) is the direct regression test for finding 2, and the probe-coroutine-on-a-single-thread-dispatcher construction is the right way to assert it — a tryLock-parked writer deadlocks the probe. Clean mutation story.
• Test 4 (cancelling a queued writer abandons its write) is the direct regression test for finding 1, and asserting process.writes == [0] after the pipe drains is exactly right — checking only at cancellation time would pass against the old code.

Gaps:

• CompletableFuture.runAsync on blocking work, still. These tasks block without a ManagedBlocker, so the common ForkJoinPool creates no compensation threads and CompletableFuture only falls back to thread-per-task at parallelism ≤ 1. On a 3-core runner (parallelism 2) test 1's permanently-blocked holder pins one worker and the six "concurrent" writers run one at a time on the other. Assertions still pass (~1.8 s against a 30 s latch) but the test stops exercising the concurrency in its name. Executors.newFixedThreadPool(writers + 1), shut down in finally, makes it independent of runner core count. Fix this →
• Test 2's ordering assertion can't fail for ordering reasons. One queued writer means [0, 1] distinguishes "wrote in order" from "didn't write at all", not from "wrote out of order". Two queued writers with a deterministic arrival gap would pin the FIFO property the comment claims.
• Nothing pins the RESOURCE_EXHAUSTED-vs-FAILED_PRECONDITION split for a waiter that acquires after exit (finding 7's second bullet). TerminalPipeFailureTest covers the uncontended exited-process case only.
• Nothing pins the require from ⬆️ actions:(deps): Bump supabase/setup-cli from 2 to 3 #1 once it's added.

I could not run :boss-app-terminal:test here — Gradle execution isn't in my allowed tools in this run, so CI's result stands.

---

Security

Nothing blocking.

• NUL check — 723df2b restores '' !in name alongside the existing '' !in value and '=' !in name, matching dev. Correct and correctly scoped; the branch's net diff against its merge base leaves validateLaunchInput untouched, which is what a revert should look like.
• Auth / approval bypass — none. session(id) performs IpcCall.current() and IpcCall.requireOwner(found.ownerInstance) before send is reached, so the queue wait happens strictly after authorization and an unauthorized caller never enqueues. ownerInstance is a val, so a 5 s suspension cannot outlive an ownership change (unlike streamOutput, which re-checks per chunk because it can).
• Secrets / logging — none. inputBusy() carries no session id and no payload; RetryInfo leaks nothing.
• Injection / path handling — untouched. validateLaunchInput, the =/NUL checks and IpcEnvironment.removeCredentials are unchanged apart from the restoration above.
• Availability — the thread-starvation vector from the last review is gone; what remains is the bounded-in-time-but-not-in-count memory point in ci: use GitHub-hosted arm64 runners instead of self-hosted labels #5, reachable only by an authenticated session owner.

---

• fix/b23-terminal-input-queue

[fluck-boss]

Re-review on 723df2b: the suspending-mutex queue resolves both earlier findings (no Dispatchers.IO pinning; cancellation abandons queued writes), the rebase-dropped NUL check is restored, CI green on ubuntu+windows, Claude review is on this head, maintainer diff read clean. Squash-merging to dev.