feat: Rust unwrap hardening

[leongdl]

Fixes: N/A — internal hardening follow-up from code review of the Rust-backed bindings

What was the problem/requirement? (What/Why)

The PyO3 bindings under rust-bindings/ contained several .unwrap() calls,
direct slice indexing, and .lock().unwrap() calls on shared mutexes. Each of
these can panic at runtime, and a panic inside a #[pyclass] method surfaces on
the Python side as an uncatchable PanicException. Worse, the session and
step-parameter-space iterators run action work on detached
std::thread::spawn threads and share state behind a Mutex: a panic on one of
those threads would poison the mutex, so every subsequent .lock().unwrap()
— including the read-only property getters called from the Python thread —
would itself panic. That permanently wedges the object: a session could get
stuck reporting "An action is already running" forever, and an iterator could
become impossible to read from.

The goal of this change is to make the bindings panic-safe: convert recoverable
failures into normal Python exceptions, and ensure that even an unexpected panic
on a background thread cannot tear down the interpreter's view of these objects.

What was the solution? (How)

Removed panic-prone patterns across the bindings and added small helpers to keep
shared state usable after a failure:

• expr/symbol_table.rs — __getitem__ and get previously did
  .into_py_any(py).unwrap(). They now propagate the conversion error with ?
  instead of panicking.

• model/step_dependency_graph.rs — replaced .unwrap() on
  self.inner.node(...) and direct self.job_steps[...] indexing with
  ok_or_else that raises a PyIndexError describing the out-of-range index.
  make_node, _nodes, and topo_sorted now return PyResult.

• model/step_param_space.rs — added a lock_recover helper
  (m.lock().unwrap_or_else(|poisoned| poisoned.into_inner())) and routed every
  iterator mutex access through it so a panic inside next() / contains() /
  validate_containment() can't poison the iterator and wedge it permanently.

• sessions/session.rs — added three helpers:

  • lock_recover — recovers a poisoned session / snapshot guard instead of
    panicking; used by every lock site (including the read-only getters).
  • run_action — the shared body for spawned action threads. It builds the
    Tokio runtime, runs the action under catch_unwind, and always returns
    the Session to the shared slot and refreshes the snapshot afterward — even
    on panic or runtime-build failure — so the session is never left
    permanently "taken".
  • failed_action_status — produces a terminal FAILED ActionStatus so that
    when an action thread can't even start (e.g. the Tokio runtime fails to
    build), the Python-side poller observes a terminal state instead of
    spinning on RUNNING forever.

  The four action entry points (enter_environment, exit_environment,
  run_task, run_subprocess) were refactored to delegate their thread bodies
  to run_action, removing the duplicated Runtime::new().unwrap() /
  manual-restore logic.

• expr/errors.rs — documentation-only: added comments describing the
  three-line, caret-annotated rendering shape of message_with_expr_prefix.

What is the impact of this change?

Failures that used to crash with an uncatchable PanicException (or silently
wedge a session/iterator) now surface as ordinary, catchable Python exceptions
(IndexError, the original conversion error, or a FAILED ActionStatus).
Sessions and iterators remain usable after a background-thread failure, so the
error can be reported through the normal channel instead of cascading. There is
no change to behavior on the success path.

How was this change tested?

See DEVELOPMENT.md for information on running tests.

• Have you run the unit tests? Yes. cargo build/clippy and the existing
  Rust + Python test suites pass; the success-path behavior of the bindings is
  unchanged.

Was this change documented?

• Are relevant docstrings in the code base updated? Yes. Each new helper
  carries a doc comment explaining the panic/poisoning scenario it guards
  against, and expr/errors.rs gained comments documenting the error-rendering
  shape.

Is this a breaking change?

No. This is purely defensive hardening of the bindings' internals. The
public Python contract is unchanged on the success path; the only observable
difference is that certain failure modes now raise catchable exceptions instead
of panicking or hanging.

Does this change impact security?

No new files/directories or permission-sensitive behavior is introduced. The
change reduces the risk of an unhandled panic crashing the interpreter, which is
a robustness improvement.

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[seant-aws]

this is good to remember
would it also be possible to add some sort of linting for rust to fail CI whenever we have a change with .unwrap() ?

[leongdl]

this is good to remember would it also be possible to add some sort of linting for rust to fail CI whenever we have a change with .unwrap() ?

Yeah we absolutely should do that. Since we have claude reviewer now, we should add an AGENTS.md, I'll set something up.

[leongdl]

The key change is on this line in the old code (repeated in each std::thread::spawn closure):

// OLD — in each spawn closure:
let rt = tokio::runtime::Runtime::new().unwrap();

And the new code in the extracted run_action helper:

// NEW:
let rt = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => { /* graceful FAILED status instead of panic */ }
};

Here's how the new tokio runtime is different:

1. new_current_thread() vs Runtime::new() (which is new_multi_thread())

• Old: Runtime::new() is shorthand for Builder::new_multi_thread().enable_all().build(). This creates a multi-threaded scheduler with a thread pool (default: one worker thread per CPU core). Each spawned action
  thread was creating an entire thread pool just to block_on a single future.
• New: Builder::new_current_thread() creates a single-threaded runtime that drives all async work on the current thread (the already-spawned std::thread). No additional worker threads are created.

2. Practical differences:

┌───────────────────────┬────────────────────────────────────────────────────────────┬───────────────────────────────────────────────────────────────────────┐
│ Aspect │ Old (Runtime::new()) │ New (new_current_thread()) │
├───────────────────────┼────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────┤
│ Threads │ Spawns N worker threads (N = CPU cores) per action │ Zero extra threads — runs on the existing spawned thread │
├───────────────────────┼────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────┤
│ Overhead │ Heavy — thread pool, work stealing, cross-thread signaling │ Lightweight — single cooperative event loop │
├───────────────────────┼────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────┤
│ Resource usage │ Could exhaust OS thread limits under high concurrency │ Minimal fd/thread pressure │
├───────────────────────┼────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────┤
│ Behavior for block_on │ Functionally identical for a single future │ Functionally identical for a single future │
├───────────────────────┼────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────┤
│ tokio::spawn inside │ Tasks can run in parallel on worker threads │ Tasks are multiplexed on the one thread (concurrent but not parallel) │
└───────────────────────┴────────────────────────────────────────────────────────────┴───────────────────────────────────────────────────────────────────────┘

3. Why it doesn't matter here:

Every call site does a single rt.block_on(session.some_async_method(...)) — there's no internal parallelism needed. The session methods are sequential async operations (spawn a subprocess, wait for it). A
current-thread runtime is semantically equivalent but uses far fewer OS resources.

4. Error handling improvement:

The old code .unwrap()'d the runtime creation — if it failed (e.g., fd exhaustion), the thread would panic silently, leaving the session permanently in the "action is running" state with no way to recover. The
new code handles the error gracefully by writing a terminal FAILED ActionStatus and returning the session to its slot.

TL;DR: The runtime switched from multi-thread (wasteful thread pool per action) to current-thread (single cooperative loop on the existing thread). For a block_on of one future this is functionally identical but
much cheaper on resources, and the error path is now graceful instead of panicking.