feat(nooa): add the sandbox process entrypoint

responses_api_agents/nooa_agent/gym_llm.py

@@ -236,6 +236,7 @@ def __init__(
         cookies: dict[str, str],
         model: str = "gym-policy",
         on_call: Callable[[GymModelCall], None] | None = None,
+        on_response: Callable[[], None] | None = None,
         sampling_overrides: dict[str, Any] | None = None,
     ) -> None:
         super().__init__(model=model)
@@ -244,6 +245,7 @@ def __init__(
         self._model_url_path = model_url_path
         self._state = state
         self._on_call = on_call
+        self._on_response = on_response
         self._sampling_overrides = dict(sampling_overrides or {})
         self._cookies = cookies
         self._calls = 0
@@ -346,6 +348,8 @@ async def _acall(
 
         # A successful retry clears only the earlier transport failure.
         self._state.fatal_error = None
+        if self._on_response is not None:
+            self._on_response()
 
         # Capture the upstream response first, but never hand a truncated tool
         # call to NOOA for execution. Its retry loop may recover with a later call.

responses_api_agents/nooa_agent/observability.py

@@ -16,7 +16,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -82,10 +82,11 @@ class GymTraceHooks:
     calls without installing global exporters or patching agent construction.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, *, on_progress: Callable[[], None] | None = None) -> None:
         self._current: ContextVar[str] = ContextVar("gym_nooa_invocation", default="root")
         self._invocations: dict[str, AgentInvocation] = {}
         self._events: list[GymModelCall | _ToolCall] = []
+        self._on_progress = on_progress
 
     def _invocation(self, identity: str) -> AgentInvocation:
         return self._invocations.setdefault(identity, AgentInvocation(invocation_id=identity))
@@ -146,6 +147,8 @@ def _finish_tool(self, event: _ToolCall, exception: BaseException | None) -> Non
             record.output = {"error": str(exception), "error_type": record.error_type}
         elif record.status == "incomplete":
             record.status = "completed"
+        if self._on_progress is not None:
+            self._on_progress()
 
     @contextmanager
     def resource_call(self, name: str, arguments: dict[str, Any]) -> Iterator[TrajectoryToolCall]:
@@ -178,7 +181,20 @@ def before_tool_execution(
             return None
         return self._start_tool(tool_name, arguments, str(kwargs.get("tool_call_id") or execution_id))
 
-    after_tool_execution = after_code_execution
+    def after_tool_execution(
+        self,
+        *,
+        context: Any,
+        result: Any,
+        exception: BaseException | None,
+        tool_name: str = "",
+        **kwargs: Any,
+    ) -> None:
+        self.after_code_execution(context=context, result=result, exception=exception, **kwargs)
+        if tool_name == "return_result" and self._on_progress is not None:
+            # Value validation is a completed tool boundary, but its arbitrary
+            # Python return value is not an external action or serialized output.
+            self._on_progress()
 
     def before_generation(self, **_: Any) -> None:
         pass
@@ -203,8 +219,9 @@ def project(
         task_id: str,
         rollout_id: str,
         default_model: str = "nooa",
+        partial: bool = False,
     ) -> tuple[AgentEpisode, TrajectoryRecord]:
-        """Project captured facts without interpreting the invocation adapter."""
+        """Project evidence, omitting pending tool outputs only in partial snapshots."""
 
         invocations = {key: record.model_copy(deep=True) for key, record in self._invocations.items()}
         output: list[Any] = []
@@ -281,6 +298,12 @@ def project(
                     )
                     invocation.conversation.append(call_item)
                     output.append(call_item)
+                if partial and record.completed_at is None:
+                    # A sibling tool may still be running at a checkpoint. Keep
+                    # its start event without inventing a completed/null output.
+                    record.output = None
+                    tools.append(record)
+                    continue
                 result_item = NeMoGymFunctionCallOutput(
                     call_id=record.tool_call_id,
                     output=_json_output(record.output),

responses_api_agents/nooa_agent/runner.py

@@ -18,7 +18,10 @@
 import asyncio
 import copy
 import inspect
+import logging
+from collections.abc import Callable
 from dataclasses import dataclass, field
+from threading import Lock
 from types import FrameType
 from typing import Any, Protocol
 from uuid import uuid4
@@ -49,6 +52,9 @@
 from responses_api_agents.nooa_agent.termination import NOOATerminationReason
 
 
+logger = logging.getLogger(__name__)
+
+
 @dataclass(slots=True)
 class NOOARunRequest:
     responses_create_params: NeMoGymResponseCreateParamsNonStreaming
@@ -140,17 +146,53 @@ def __init__(
         server_client: GymModelClient,
         model_server_name: str,
         max_policy_calls: int | None,
+        on_progress: Callable[[NOOARunResult], None] | None = None,
     ) -> None:
         self._invocation = invocation
         self._server_client = server_client
         self._model_server_name = model_server_name
         self._max_policy_calls = max_policy_calls
+        self._on_progress = on_progress
         self._agent_class, self._invocation_adapter = validate_invocation(invocation)
         _validate_agent_llm_bindings(self._agent_class)
 
     async def run(self, request: NOOARunRequest) -> NOOARunResult:
         state = RolloutLLMState(max_policy_calls=self._max_policy_calls)
-        trace = GymTraceHooks()
+        progress_lock = Lock()
+
+        def checkpoint() -> None:
+            if self._on_progress is None:
+                return
+            # NOOA's synchronous callbacks do not yield to sibling async calls.
+            # Keep projection and persistence in one critical section as well,
+            # so concurrent callbacks cannot publish older snapshots last. Publish
+            # before execution resumes: a queued write could lose this boundary
+            # if the worker is killed immediately after the callback returns.
+            with progress_lock:
+                try:
+                    episode, trajectory = trace.project(
+                        create_params=request.responses_create_params,
+                        state=state,
+                        task_id=request.task_id,
+                        rollout_id=request.rollout_id,
+                        default_model=self._model_server_name,
+                        partial=True,
+                    )
+                    self._on_progress(
+                        NOOARunResult(
+                            episode=episode,
+                            return_value=None,
+                            model_cookies=dict(request.model_cookies),
+                            resource_cookies=dict(request.resource_cookies),
+                            trajectory=trajectory,
+                        )
+                    )
+                except Exception:
+                    # Evidence persistence is best effort. NOOA swallows tool
+                    # hook errors, so report all checkpoint failures consistently.
+                    logger.exception("Failed to persist NOOA partial checkpoint")
+
+        trace = GymTraceHooks(on_progress=checkpoint)
         sampling_overrides = request.responses_create_params.model_dump(
             include={"temperature", "top_p", "max_output_tokens"}, exclude_unset=True, exclude_none=True
         )
@@ -161,6 +203,7 @@ async def run(self, request: NOOARunRequest) -> NOOARunResult:
             state=state,
             cookies=request.model_cookies,
             on_call=trace.on_model_call,
+            on_response=checkpoint,
             sampling_overrides=sampling_overrides,
         )
         dispatcher = ResourceToolDispatcher(