fix(agent): wait for exec sessions without polling (#5526)

nanobot/agent/tools/exec_session.py

@@ -22,7 +22,8 @@ from nanobot.agent.tools.schema import (
 DEFAULT_YIELD_MS = 1000
 MAX_YIELD_MS = 30_000
 DEFAULT_WAIT_FOR_MS = 10_000
-MAX_WAIT_FOR_MS = 120_000
+DEFAULT_UNTIL_EXIT_MS = 600_000
+MAX_WAIT_FOR_MS = 600_000
 DEFAULT_MAX_OUTPUT_CHARS = 10_000
 MAX_OUTPUT_CHARS = 50_000
 OUTPUT_DRAIN_GRACE_S = 0.1
@@ -495,51 +496,39 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
 
 @tool_parameters(
     tool_parameters_schema(
-        session_id=StringSchema("Session id returned by exec when yield_time_ms is used."),
-        chars=StringSchema(
-            "Bytes/text to write to stdin. Omit or pass an empty string to only poll recent output.",
+        session_id=StringSchema("Session ID returned by exec."),
+        input=StringSchema(
+            "Text to send to stdin; omit to poll output.",
             nullable=True,
         ),
         close_stdin=BooleanSchema(
-            description="Close stdin after writing chars. Useful for commands waiting for EOF.",
+            description="Close stdin after sending input.",
             default=False,
         ),
         terminate=BooleanSchema(
-            description="Terminate the running exec session.",
+            description="Terminate the session; use alone.",
             default=False,
         ),
-        yield_time_ms=IntegerSchema(
-            description="Milliseconds to wait before returning recent output (default 1000, max 30000).",
-            minimum=0,
-            maximum=MAX_YIELD_MS,
-        ),
         wait_for=StringSchema(
-            "Optional text to wait for in output before returning. "
-            "Useful for interactive commands and dev servers.",
+            "Return when this text appears in output.",
+            min_length=1,
             nullable=True,
         ),
-        wait_timeout_ms=IntegerSchema(
-            description="Maximum milliseconds to wait for wait_for text (default 10000, max 120000).",
+        until_exit=BooleanSchema(
+            description="Wait for the process to exit.",
+            default=False,
+        ),
+        timeout_ms=IntegerSchema(
+            description="Maximum wait: 1s normally, 10s for wait_for, 10m for until_exit.",
             minimum=0,
             maximum=MAX_WAIT_FOR_MS,
             nullable=True,
         ),
-        max_output_chars=IntegerSchema(
-            description="Maximum output characters to return from this poll (default 10000, max 50000).",
-            minimum=1000,
-            maximum=MAX_OUTPUT_CHARS,
-        ),
-        max_output_tokens=IntegerSchema(
-            description="Compatibility alias for max_output_chars. The current runtime uses a character budget.",
-            minimum=1000,
-            maximum=MAX_OUTPUT_CHARS,
-            nullable=True,
-        ),
         required=["session_id"],
     )
 )
-class WriteStdinTool(Tool):
-    """Write to or poll a running exec session."""
+class ExecSessionTool(Tool):
+    """Interact with or wait for a running exec session."""
 
     _scopes = {"core", "subagent"}
     config_key = "exec"
@@ -571,98 +560,103 @@ class WriteStdinTool(Tool):
 
     @property
     def name(self) -> str:
-        return "write_stdin"
+        return "exec_session"
 
     @property
     def description(self) -> str:
-        return (
-            "Interact with a running exec session created by exec with "
-            "yield_time_ms. Use chars='' to poll without writing, chars to send "
-            "stdin, close_stdin=true to send EOF, or terminate=true to stop the "
-            "process. Use wait_for with wait_timeout_ms for dev servers, test "
-            "watchers, and prompts where you need to wait for expected output. "
-            "Do not use this to start new commands; start them with exec."
-        )
+        return "Manage a session returned by exec."
 
     async def execute(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         session_id: str,
-        chars: str | None = None,
+        input: str | None = None,
         close_stdin: bool = False,
         terminate: bool = False,
-        yield_time_ms: int | None = None,
         wait_for: str | None = None,
-        wait_timeout_ms: int | None = None,
-        max_output_chars: int | None = None,
-        max_output_tokens: int | None = None,
+        until_exit: bool = False,
+        timeout_ms: int | None = None,
         **kwargs: Any,
     ) -> str:
         try:
-            if max_output_chars is None:
-                max_output_chars = max_output_tokens
-            output_limit = clamp_session_int(
-                max_output_chars,
-                DEFAULT_MAX_OUTPUT_CHARS,
-                1000,
-                MAX_OUTPUT_CHARS,
-            )
-            if wait_for:
-                return await self._wait_for_output(
-                    session_id=session_id,
-                    chars=chars,
-                    close_stdin=close_stdin,
-                    terminate=terminate,
-                    wait_for=wait_for,
-                    wait_timeout_ms=clamp_session_int(
-                        wait_timeout_ms,
-                        DEFAULT_WAIT_FOR_MS,
-                        0,
-                        MAX_WAIT_FOR_MS,
-                    ),
-                    max_output_chars=output_limit,
+            if wait_for == "":
+                return ToolResult.error("Error: wait_for must not be empty.")
+            if wait_for is not None and until_exit:
+                return ToolResult.error(
+                    "Error: wait_for and until_exit are mutually exclusive."
                 )
-            poll = await self._manager.write(
+            if terminate:
+                if any(
+                    (
+                        input is not None,
+                        close_stdin,
+                        wait_for is not None,
+                        until_exit,
+                        timeout_ms is not None,
+                    )
+                ):
+                    return ToolResult.error("Error: terminate must be used alone.")
+                poll = await self._manager.write(
+                    session_id=session_id,
+                    chars=None,
+                    close_stdin=False,
+                    terminate=True,
+                    yield_time_ms=0,
+                    max_output_chars=DEFAULT_MAX_OUTPUT_CHARS,
+                    owner_session_key=current_request_session_key(),
+                )
+                result = format_session_poll(session_id, poll)
+                return ToolResult.error(result) if poll.timed_out else result
+
+            default_timeout_ms = (
+                DEFAULT_UNTIL_EXIT_MS
+                if until_exit
+                else DEFAULT_WAIT_FOR_MS
+                if wait_for is not None
+                else DEFAULT_YIELD_MS
+            )
+            return await self._wait(
                 session_id=session_id,
-                chars=chars,
+                input=input,
                 close_stdin=close_stdin,
-                terminate=terminate,
-                yield_time_ms=clamp_session_int(yield_time_ms, DEFAULT_YIELD_MS, 0, MAX_YIELD_MS),
-                max_output_chars=output_limit,
-                owner_session_key=current_request_session_key(),
+                wait_for=wait_for,
+                until_exit=until_exit,
+                timeout_ms=clamp_session_int(
+                    timeout_ms,
+                    default_timeout_ms,
+                    0,
+                    MAX_WAIT_FOR_MS,
+                ),
             )
-            result = format_session_poll(session_id, poll)
-            return ToolResult.error(result) if poll.timed_out else result
         except KeyError:
             return ToolResult.error(f"Error: exec session not found: {session_id!r}")
         except Exception as exc:
-            return ToolResult.error(f"Error writing to exec session: {exc}")
+            return ToolResult.error(f"Error managing exec session: {exc}")
 
-    async def _wait_for_output(
+    async def _wait(
         self,
         *,
         session_id: str,
-        chars: str | None,
+        input: str | None,
         close_stdin: bool,
-        terminate: bool,
-        wait_for: str,
-        wait_timeout_ms: int,
-        max_output_chars: int,
+        wait_for: str | None,
+        until_exit: bool,
+        timeout_ms: int,
     ) -> str:
-        deadline = time.monotonic() + (wait_timeout_ms / 1000)
-        aggregate = _BoundedOutputBuffer(max_output_chars)
+        deadline = time.monotonic() + (timeout_ms / 1000)
+        aggregate = _BoundedOutputBuffer(DEFAULT_MAX_OUTPUT_CHARS)
         upstream_truncated = 0
         search_overlap = ""
         first = True
-        poll: _SessionPoll | None = None
+        matched = False
 
         while True:
             remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
-            step_ms = min(500, remaining_ms)
+            step_ms = min(MAX_YIELD_MS if until_exit else 500, remaining_ms)
             poll = await self._manager.write(
                 session_id=session_id,
-                chars=chars if first else None,
+                chars=input if first else None,
                 close_stdin=close_stdin if first else False,
-                terminate=terminate if first else False,
+                terminate=False,
                 yield_time_ms=step_ms,
                 max_output_chars=MAX_OUTPUT_CHARS,
                 owner_session_key=current_request_session_key(),
@@ -671,20 +665,25 @@ class WriteStdinTool(Tool):
             upstream_truncated += poll.truncated_chars
             if poll.output:
                 aggregate.append(poll.output)
-                searchable = search_overlap + poll.output
-                if wait_for in searchable:
-                    poll.output, aggregate_truncated = aggregate.drain()
-                    poll.truncated_chars = upstream_truncated + aggregate_truncated
-                    result = format_session_poll(session_id, poll)
-                    return ToolResult.error(result) if poll.timed_out else result
-                overlap_chars = max(0, len(wait_for) - 1)
-                search_overlap = searchable[-overlap_chars:] if overlap_chars else ""
-            if poll.done or remaining_ms <= 0:
+                if wait_for is not None:
+                    searchable = search_overlap + poll.output
+                    matched = wait_for in searchable
+                    overlap_chars = len(wait_for) - 1
+                    search_overlap = searchable[-overlap_chars:] if overlap_chars else ""
+
+            expired = time.monotonic() >= deadline
+            has_activity = wait_for is None and not until_exit and bool(poll.output)
+            if poll.done or matched or has_activity or expired:
                 poll.output, aggregate_truncated = aggregate.drain()
                 poll.truncated_chars = upstream_truncated + aggregate_truncated
                 result = format_session_poll(session_id, poll)
-                if wait_for not in poll.output:
+                if wait_for is not None and not matched:
                     result += f"\nWait target not observed: {wait_for!r}"
+                elif until_exit and not poll.done:
+                    result += (
+                        f"\nWait timed out after {timeout_ms / 1000:g}s; "
+                        "session remains active."
+                    )
                 return ToolResult.error(result) if poll.timed_out else result
 
 
@@ -722,12 +721,7 @@ class ListExecSessionsTool(Tool):
 
     @property
     def description(self) -> str:
-        return (
-            "List active long-running exec sessions, including session_id, cwd, "
-            "elapsed time, idle time, remaining timeout, and command preview. "
-            "Use this to recover a session_id after context shifts before "
-            "polling, writing stdin, or terminating with write_stdin."
-        )
+        return "List active exec sessions."
 
     @property
     def read_only(self) -> bool:

nanobot/agent/tools/registry.py

@@ -70,7 +70,7 @@ class ToolRegistry:
 
     def has(self, name: str) -> bool:
         """Check if a tool is registered."""
-        return name in self._tools
+        return self.get(name) is not None
 
     @staticmethod
     def _schema_name(schema: dict[str, Any]) -> str:
@@ -113,7 +113,7 @@ class ToolRegistry:
         params: Any,
     ) -> tuple[Tool | None, Any, str | None]:
         """Resolve, cast, and validate one tool call."""
-        tool = self._tools.get(name)
+        tool = self.get(name)
         if not tool:
             suggestion = self._suggest_name(str(name))
             hint = f" Did you mean '{suggestion}'? Tool names must match exactly." if suggestion else ""
@@ -209,4 +209,4 @@ class ToolRegistry:
         return len(self._tools)
 
     def __contains__(self, name: str) -> bool:
-        return name in self._tools
+        return self.has(name)

nanobot/agent/tools/shell.py

@@ -122,55 +122,37 @@ class _PreparedCommand:
         working_dir=StringSchema("Optional working directory for the command"),
         workdir=StringSchema("Compatibility alias for working_dir"),
         timeout=IntegerSchema(
-            description=(
-                "Timeout in seconds. Increase for long-running commands "
-                "like compilation or installation (default 60, max 600)."
-            ),
+            description="Hard timeout in seconds (default 60, max 600).",
             minimum=1,
             maximum=600,
         ),
         shell=StringSchema(
             (
-                "Override the Windows shell only when needed. Omit to use "
-                "PowerShell by default (pwsh when available, else powershell). "
-                "Pass 'cmd' only for cmd.exe syntax or cmd built-ins."
+                "Shell override; omit for PowerShell, or pass 'cmd' for cmd.exe."
                 if _IS_WINDOWS
-                else "Override the Unix shell only when needed. Omit to use "
-                "bash by default. Pass 'sh' for POSIX sh or 'zsh' for "
-                "zsh-specific syntax."
+                else "Shell override; omit for bash, or pass 'sh' or 'zsh'."
             ),
             nullable=True,
         ),
         login=BooleanSchema(
-            description="Whether to run bash/zsh with login shell semantics (default false).",
+            description="Run bash/zsh as a login shell.",
             default=False,
             nullable=True,
         ),
         yield_time_ms=IntegerSchema(
-            description=(
-                "Optional milliseconds to wait before returning output. "
-                "When set, a still-running command returns a session_id that "
-                "can be polled or written to with write_stdin. Omit this field "
-                "to keep one-shot exec behavior."
-            ),
+            description="Return after this many milliseconds if still running; omit to wait for exit.",
             minimum=0,
             maximum=MAX_YIELD_MS,
             nullable=True,
         ),
         max_output_chars=IntegerSchema(
-            description=(
-                "Maximum output characters to return when yield_time_ms is used "
-                "(default 10000, max 50000)."
-            ),
+            description="Session output limit in characters (default 10000, max 50000).",
             minimum=1000,
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
         ),
         max_output_tokens=IntegerSchema(
-            description=(
-                "Compatibility alias for max_output_chars. The current runtime "
-                "uses a character budget."
-            ),
+            description="Compatibility alias for max_output_chars.",
             minimum=1000,
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
@@ -283,26 +265,7 @@ class ExecTool(Tool):
 
     @property
     def description(self) -> str:
-        platform_note = (
-            "On Windows, use PowerShell syntax by default; pass shell='cmd' "
-            "only for cmd-specific commands. "
-            if _IS_WINDOWS
-            else "On Unix, commands run through bash by default; pass shell='sh' "
-            "or shell='zsh' when needed. "
-        )
-        return (
-            "Execute a shell command and return its output. "
-            "Use this for tests, builds, package commands, git commands, and "
-            "other process execution. Prefer read_file/find_files/grep for "
-            "inspection and apply_patch/write_file/edit_file for file changes "
-            "instead of cat, shell find/grep, echo, or sed. "
-            "Use -y or --yes flags to avoid interactive prompts. "
-            f"{platform_note}"
-            "For long-running or interactive commands, pass yield_time_ms; "
-            "if the command keeps running, exec returns a session_id that can "
-            "be polled or written to with write_stdin. Output is truncated at "
-            "10 000 chars; timeout defaults to 60s."
-        )
+        return "Execute a shell command."
 
     @property
     def exclusive(self) -> bool:

nanobot/session/manager.py

@@ -82,6 +82,111 @@ def _json_object(value: object) -> dict[str, Any]:
     return cast(dict[str, Any], value)
 
 
+# TODO(0.3.2): Remove the write_stdin replay migration after 0.3.1.
+def _migrate_legacy_exec_arguments(container: dict[str, Any]) -> bool:
+    raw_arguments = cast(object, container.get("arguments"))
+    encoded = isinstance(raw_arguments, str)
+    if encoded:
+        try:
+            decoded: object = json.loads(raw_arguments)
+        except json.JSONDecodeError:
+            return False
+    else:
+        decoded = raw_arguments
+    if not isinstance(decoded, dict):
+        return False
+
+    arguments = cast(dict[str, Any], decoded)
+    changed = False
+    if "chars" in arguments:
+        if "input" not in arguments:
+            arguments["input"] = arguments["chars"]
+        arguments.pop("chars")
+        changed = True
+
+    wait_key = (
+        "wait_timeout_ms"
+        if arguments.get("wait_for") or arguments.get("until_exit")
+        else "yield_time_ms"
+    )
+    if "timeout_ms" not in arguments and wait_key in arguments:
+        arguments["timeout_ms"] = arguments[wait_key]
+    for key in ("yield_time_ms", "wait_timeout_ms", "max_output_chars", "max_output_tokens"):
+        if key in arguments:
+            arguments.pop(key)
+            changed = True
+
+    if changed:
+        container["arguments"] = (
+            json.dumps(arguments, ensure_ascii=False, separators=(",", ":"))
+            if encoded
+            else arguments
+        )
+    return changed
+
+
+def _migrate_legacy_exec_tool_call(value: object) -> bool:
+    if not isinstance(value, dict):
+        return False
+    tool_call = cast(dict[str, Any], value)
+    function_value = cast(object, tool_call.get("function"))
+    function = (
+        cast(dict[str, Any], function_value)
+        if isinstance(function_value, dict)
+        else tool_call
+    )
+    name = function.get("name")
+    if name not in {"write_stdin", "exec_session"}:
+        return False
+
+    changed = name == "write_stdin"
+    if changed:
+        function["name"] = "exec_session"
+    return _migrate_legacy_exec_arguments(function) or changed
+
+
+def _migrate_legacy_exec_message(message: dict[str, Any]) -> bool:
+    changed = False
+    if message.get("name") == "write_stdin":
+        message["name"] = "exec_session"
+        changed = True
+    tool_calls = cast(object, message.get("tool_calls"))
+    if isinstance(tool_calls, list):
+        for tool_call in cast(list[object], tool_calls):
+            changed = _migrate_legacy_exec_tool_call(tool_call) or changed
+    return changed
+
+
+def _migrate_legacy_exec_session_records(
+    messages: list[dict[str, Any]],
+    metadata: dict[str, Any],
+) -> bool:
+    changed = False
+    for message in messages:
+        changed = _migrate_legacy_exec_message(message) or changed
+
+    checkpoint_value = cast(object, metadata.get(_RUNTIME_CHECKPOINT_KEY))
+    if not isinstance(checkpoint_value, dict):
+        return changed
+    checkpoint = cast(dict[str, Any], checkpoint_value)
+    assistant = cast(object, checkpoint.get("assistant_message"))
+    if isinstance(assistant, dict):
+        changed = _migrate_legacy_exec_message(cast(dict[str, Any], assistant)) or changed
+    pending = cast(object, checkpoint.get("pending_tool_calls"))
+    if isinstance(pending, list):
+        for tool_call in cast(list[object], pending):
+            changed = _migrate_legacy_exec_tool_call(tool_call) or changed
+    completed = cast(object, checkpoint.get("completed_tool_results"))
+    if isinstance(completed, list):
+        for result in cast(list[object], completed):
+            if isinstance(result, dict):
+                result_data = cast(dict[str, Any], result)
+                if result_data.get("name") == "write_stdin":
+                    result_data["name"] = "exec_session"
+                    changed = True
+    return changed
+
+
 def _is_provider_state_record_line(line: str) -> bool:
     """Recognize the canonical private record without decoding its opaque payload."""
     return _PROVIDER_STATE_RECORD_PREFIX_RE.match(line) is not None
@@ -1085,6 +1190,8 @@ class JsonlSessionStore:
                 provider_state=provider_state,
             )
             self._overlay_runtime_checkpoint_unlocked(session, path)
+            if _migrate_legacy_exec_session_records(session.messages, session.metadata):
+                session.provider_state = None
             return session
         except _SESSION_DATA_ERRORS as e:
             logger.warning("Failed to load session {}: {}", key, e)
@@ -1180,6 +1287,8 @@ class JsonlSessionStore:
                 provider_state=provider_state,
             )
             self._overlay_runtime_checkpoint_unlocked(session, path)
+            if _migrate_legacy_exec_session_records(session.messages, session.metadata):
+                session.provider_state = None
             return session
         except _SESSION_DATA_ERRORS as e:
             logger.warning("Repair failed for session {}: {}", key, e)
@@ -1454,6 +1563,7 @@ class JsonlSessionStore:
                         continue
                     else:
                         messages.append(data)
+            _migrate_legacy_exec_session_records(messages, metadata)
             return {
                 "key": stored_key or key,
                 "created_at": created_at,

nanobot/templates/agent/tool_contract.md

@@ -48,13 +48,9 @@
 
 ## Process Execution
 
-- Use `exec` for tests, builds, package commands, git commands, and other process execution.
-- Prefer dedicated file/search tools over `cat`, shell `find`, shell `grep`, `sed`, or `echo` for ordinary workspace inspection and edits.
-- Use non-interactive flags such as `-y` or `--yes` when available.
-- Commands have a configurable timeout (default 60s), dangerous commands are blocked, and output is truncated.
-- For long-running or interactive commands, pass `yield_time_ms`; if the process keeps running, continue with `write_stdin`.
-- Use `write_stdin` to poll, provide stdin, close stdin, wait for expected output with `wait_for`, or terminate an existing exec session.
-- Use `list_exec_sessions` to recover active session IDs after context shifts.
+- Use `exec` for processes, not file inspection or editing.
+- For interaction or early output, set `yield_time_ms` and continue with `exec_session` (`until_exit=true` when no further input is needed).
+- Use `list_exec_sessions` to recover session IDs.
 
 ## CLI App Attachments