fix(flowing): close sessions a flow can no longer reach

[futrime]

Summary

The flow API has no Session.close, and the engine closed a session only as the flow call that opened it ended, holding every SessionView strongly until then. A loop that opens a fresh session a round (ralph_loop, flame_chase, rlar's reviewer, lane runtimes, which can run for days) kept all of them open. With stream harnesses that is one CLI process per round.

Session lifetime (viewing.py, engine.py)

• The engine now holds a SessionView only weakly, through an Opened record (a weakref.ref subclass). The call's cleanup and the agent's hooks keep the record instead of the view.
• When a flow lets go of a view, the session is closed on the run's loop. This works whichever thread dropped the last reference or ran the collection that found it. The close runs in an eagerly started task, and both the call's cleanup and the run's cleanup wait for it.
• The next spawn first closes whatever was let go of, so a flow that never yields to the loop (a fake one, say) also keeps few sessions open.
• Each session is closed exactly once, whether it is let go of first or its call ends first. A session handed up to a caller still closes when the call that opened it ends.
• Call.res is keyed by id, so a session that closes early leaves it.
• A hook that fires for a session the flow let go of gets a stand-in view that is already closed.
• A fork keeps the session it was forked from open until the fork's own first turn succeeds, since that turn is where a harness actually cuts the fork.
• A hook that fails between turns no longer holds its session in a reference cycle through the hook's frame.
• steer now refuses a session that is closed.

Hard deadlines under a sooner graceful one (a follow-up bug from the coordinator)

• Call.limits now also returns the hard deadline that a sooner graceful deadline hides from the driver. AgentView.run arms a timer for it that interrupts the turn and raises DurationExceeded. Only a SessionError from the interrupted turn becomes DurationExceeded; a turn that answered in time keeps its answer.
• A turn that is hard for its cost or tokens is now told the hard deadline, not the graceful one.
• The SPI is unchanged.
• A graceful deadline now sends its cancel from the loop. Before, a flow that returned just as its last turn ended passed a bare CancelledError up to its caller.

Fakes (fakes.py)

• FakeAgentDriver.live and .peak count how many sessions are open now, and the most open at once.
• A fake turn that its reply holds open (until_steered()) is cut off at a hard deadline, as a real turn is.
• FakeEnvDriver releases temporary-copy holds when the run that took them closes (or when the root fake closes), and keeps the copies. A run resumed on the same fake gets its copy back instead of TempCloneBusy.

The spec (specs/runtime/flowing.md, the cleanup section only), the Session docstring, and the docs (reference/flows.md, features/budgets.md, weaver/testing-flows.md) are updated to match.

Tests

• tests/unit/flows/test_engine_sessions.py (new):
  • 10,000 rounds with a fresh session each, with instant and yielding replies: never more than 2 sessions open at once, and every session closed at the end. The same holds for gathered batches (at most one batch open) and for 10 loops gathered (at most 2 open per loop).
  • An on-disk flow run through run_flow.
  • Sessions kept in a local variable, a list or a dict stay open.
  • Forks: one outlives the session it was forked from, one whose first turn failed still holds that session, and a chain of forks keeps few sessions open.
  • Exactly-once close: when the view goes as the call ends, and when a close is still under way as the call ends.
  • A session let go of on another thread (released, or found by a collection) is closed on the loop; the check runs with the loop in debug mode.
  • A hook sees a stand-in for a session the flow let go of.
  • A hook that fails between turns no longer keeps its session open.
• test_engine_budget.py:
  • Hard deadline cases: the turn's own hard deadline with a sooner graceful one from the run or from its flow; a sooner hard deadline with a later graceful one; the flow's hard deadline with a sooner graceful one from the run.
  • A turn that is hard for its cost is told the hard deadline.
  • A flow that returns as its graceful deadline lets its turn finish gets its result.
  • Timers are cleaned up after 10,000 turns.
• test_engine_resume.py and test_fakes.py: a resumed run takes its copy again, and fake holds are released when the fake closes.
• test_engine_scale.py: a new micro-benchmark compares spawn + turn + close per round, letting go of each session versus keeping all of them. Result: 20.8 µs vs 20.0 µs (1.04×). The existing thresholds are unchanged and still pass. Per-call engine cost is the same as the base (1.88 µs vs 1.89 µs).
• Every new test fails on the base branch and passes here.
• uv run pre-commit run --all-files passes. uv run pytest passes (4149 passed, 110 skipped). The system tier ran without --run-agents, so no real harnesses were driven.

E2E: an on-disk flow run through run_flow on the fake kit, 1,000 rounds with a fresh session each: at most 2 sessions open at once, 0 open at the end, all 1,000 closed, in about 38 ms. On the base branch the peak is 1,000.

🤖 Generated with Claude Code