Fix cancelled ingest waiters wedging sessions

[SabhyaC26]

Related issue

N/A

Summary

• Replace the runner's custom ticket/condition ingest gate with a per-session FIFO asyncio.Lock, so cancelled waiters are removed instead of permanently blocking the queue.
• Apply the same gate to direct message intake and buffered-turn continuation, with a regression test for the cancelled-waiter sequence.

ELI5: A cancelled request used to keep its number in line forever. The standard lock drops cancelled waiters, so the next message can move forward.

flowchart LR
    A[Active ingest] --> B[Cancelled waiter]
    B --> C[Later message]
    B -. removed from lock queue .-> C

Loading

Test Plan

• pytest -q tests/runner/test_app_sessions_native_workflow_messages.py — 25 passed.
• pre-commit run --files omnigent/runner/app.py tests/runner/test_app_sessions_native_workflow_messages.py — all applicable hooks passed, including Ruff and Pyrefly.
• Started a disposable live server and host from this branch, then used the in-app Browser UI to launch a host-backed Codex runner. Sent a command containing sleep 3, queued a follow-up during the active turn, and sent another message afterward; the session returned LIVE_READY, FIRST_DONE, SECOND_DONE, and THIRD_DONE without wedging.

Demo

N/A — backend concurrency fix with no visual change.

Type of change

• Bug fix
• Feature
• UI / frontend change
• Refactor / chore
• Docs
• Test / CI
• Breaking change

Test coverage

• Unit tests added / updated
• Integration tests added / updated
• E2E tests added / updated
• Manual verification completed
• Existing tests cover this change
• Not applicable

Coverage notes

The new regression cancels a waiter parked behind a slow ingest and verifies the next message completes. The existing ordering test confirms FIFO delivery is preserved. Manual verification exercised the full browser → server → host → runner path.

Changelog

Sessions continue accepting messages when a queued wake request is cancelled.

[omnigent-ci]

Polly AI Review

Blocking issues

None.

Security vulnerabilities

None. No new dependencies, extras, or lockfile changes; no change to any auth/trust boundary.

Non-blocking notes

• FIFO is preserved by the swap. The original bug is real: the condition/ticket gate consumed a monotonic _seq per caller, but a waiter cancelled inside await _cond.wait() never reached the finally that advances _ingest_now_serving, so now_serving could never catch up to a later waiter's seq — permanently wedging the session. asyncio.Lock maintains a FIFO waiter queue and drops cancelled waiters from it, and async with guarantees release on exception/cancellation, so the wedge is eliminated while ordering is retained. Order is still established synchronously at function entry (no await precedes async with _ingest_lock), matching the old ticket-at-entry semantics. Good.
• No reentrancy/deadlock risk. Both call sites of _check_and_start_next_turn and all _run_turn_bg invocations are scheduled via asyncio.create_task rather than awaited while the lock is held, so the critical section never re-enters the same session's lock. The gated file-meta fetch (_resolve_forwarded_message_content) is awaited inside the lock, which is intended — that's what serializes ingest.
• Minor: _ingest_locks.setdefault(session_id, asyncio.Lock()) eagerly allocates a Lock on every call and discards it when one already exists. Harmless (single-threaded, setdefault is atomic with no intervening await), just a tiny allocation churn; not worth changing.
• Cleanup in delete_session correctly drops the single _ingest_locks entry replacing the three former dicts, and no other module referenced the removed _ingest_next_seq / _ingest_now_serving / _ingest_cond names.
• The regression test faithfully reproduces the scenario (park a waiter behind a gated ingest, cancel it, confirm a later message still reaches 202) and would deadlock/timeout against the old code, so it genuinely guards the fix.

Summary

Solid, well-scoped concurrency fix. Replacing the hand-rolled ticket/condition gate — whose finally was skipped on a cancelled waiter, stranding the sequence counter — with a per-session asyncio.Lock is the right call: it drops cancelled waiters automatically, releases deterministically via async with, and preserves FIFO ordering. The change is applied consistently to both ingest paths, cleanup is updated, and the added regression test targets the exact failure mode. No blocking or security concerns; ready to merge.

---

_{Automated review by Polly · workflow run}