diff --git a/docs/agent.md b/docs/agent.md index b959135..6743e96 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -136,6 +136,12 @@ ### Sub-agent event forwarding When `spawn_agent` runs a subagent, its events arrive through `current_event_sink`. The parent drains the queue in real time, yielding subagent events marked with `is_subagent=True`. +### Background tools +`ToolExecutor._execute_one` intercepts `background: true` in the arguments of a tool from `settings.backgroundable_tools` (`terminal`, `ssh_exec`, `peer`, `spawn_agent`, `code_exec` — `terminal` action `open` is never hijacked). The tool's `execute()` is submitted to the `TaskManager` ([`tasks.md`](tasks.md)) under a per-task ring event sink and per-task stop event — a session Stop does NOT affect background tasks. The foreground loop immediately receives a synthetic `ToolResult` (`task_id`, `args_summary`, hint to use the `tasks` tool) and the turn continues normally. Caps: per-session, global, spawn-specific and rate-limited — exceeding a cap runs the tool inline as a failed result instead of detaching. + +### Parallel tool batches +When `profile.parallel_tool_calls ?? settings.parallel_tool_calls` (default off) is enabled and the model returns several tool calls in one response, `_execute_tools_parallel` starts them all together: `ToolStarted` for every call up-front, each call's events multiplexed through a tagged shared queue, and results/messages appended in call order after the whole batch settles — one `save()` per batch. A mid-batch Stop cancels the remaining calls and synthesizes `is_context=False` stopped results. A session loaded with a batch interrupted before its results is repaired on load (`pg_session_store.repair_dangling_tool_calls`). + ### Cooperative stop Stop is signalled via `current_stop_event` (an `asyncio.Event`). Checked before each LLM call, during streaming, and after tool execution. Never use `task.cancel()` — it corrupts WebSocket state. diff --git a/docs/api.md b/docs/api.md index 13aa0c7..d4216b7 100644 --- a/docs/api.md +++ b/docs/api.md @@ -992,6 +992,33 @@ --- +#### `task_update` +```json +{ + "type": "task_update", + "task_id": "bt-1a2b3c4d", + "session_id": "...", + "tool": "terminal", + "status": "running", + "result_preview": "", + "parent_tool_call_id": "", + "started_at": "2026-04-30T10:00:00+00:00", + "finished_at": null, + "subagent_tokens": null +} +``` +A background task (a tool call detached with `background: true`, see `docs/tasks.md`) changed state. `status` is `running | completed | failed | cancelled`. Out-of-band: NOT part of the run's replay buffer — a reconnecting client learns task state via the `tasks` tool/notes, not replay. `parent_tool_call_id` (when non-empty) binds the update to the tool card that started the task, so the client can attach sub-agent progress to it. + +--- + +#### `message_queued` +```json +{ "type": "message_queued", "position": 1, "queue_len": 1, "max": 5, "dropped_total": 0 } +``` +The user's message arrived while a run was active and was queued instead of erroring. It will execute after the current run finishes. `queue_len` counts queued messages, `max` is `MESSAGE_QUEUE_MAX`, `dropped_total` grows when the queue was full and the oldest message was dropped. + +--- + #### `session_sync` ```json { "type": "session_sync", "session_id": "...", "profile_id": "..." } diff --git a/docs/config.md b/docs/config.md index f48d87a..2ac39ab 100644 --- a/docs/config.md +++ b/docs/config.md @@ -215,6 +215,24 @@ The `_load_persona_from_file` validator reads the file on startup if `NAVI_PERSONA` is empty and `NAVI_PERSONA_FILE` is set. +## Parallelism and background tasks + +| Variable | Type | Default | Description | +|---|---|---|---| +| `TASKS_MAX_PER_SESSION` | int | `5` | Concurrent running background tasks per session | +| `TASKS_MAX_GLOBAL` | int | `20` | Concurrent running background tasks server-wide | +| `TASKS_MAX_SPAWN` | int | `2` | Concurrent background `spawn_agent` per session | +| `TASKS_TTL_SEC` | int | `3600` | Finished-task retention before reaping | +| `TASKS_EVENT_BUFFER_SIZE` | int | `50` | Per-task event ring (spawn tasks use 200) | +| `TASKS_RATE_LIMIT` | int | `10` | Max task spawns per 5 min per session | +| `TASK_NOTES_MAX_PENDING` | int | `20` | Pending completion notes per session | +| `TASK_NOTES_PER_TURN` | int | `5` | Max notes coalesced into one turn injection | +| `MESSAGE_QUEUE_MAX` | int | `5` | User messages queued while a run is active | +| `BACKGROUNDABLE_TOOLS` | str | `terminal,ssh_exec,peer,spawn_agent,code_exec` | Tools accepting `"background": true` | +| `PARALLEL_TOOL_CALLS` | bool | `false` | Execute a multi-tool-call batch concurrently (per-profile override: `AgentProfile.parallel_tool_calls`) | + +See [`docs/tasks.md`](tasks.md) for the full mechanism. + ## Example `.env` ```dotenv diff --git a/docs/index.md b/docs/index.md index 2930f2e..cfb8ed0 100644 --- a/docs/index.md +++ b/docs/index.md @@ -32,6 +32,7 @@ | [`sessions.md`](sessions.md) | Session model, dual-buffer design, context compression | | [`store.md`](store.md) | KV store — per-session/per-user key-value persistence | | [`recall.md`](recall.md) | Scheduled callbacks — headless recall system | +| [`tasks.md`](tasks.md) | Background tasks — detach tool calls, TaskManager, completion notes, message queue | | [`websocket.md`](websocket.md) | WebSocket protocol — all events, stop mechanism | | [`push.md`](push.md) | PWA — install, offline shell, web push (VAPID, trigger rule) | | [`profiles.md`](profiles.md) | Profiles, system prompts, persona, profile switching | diff --git a/docs/tasks.md b/docs/tasks.md new file mode 100644 index 0000000..0113c0f --- /dev/null +++ b/docs/tasks.md @@ -0,0 +1,152 @@ +# Background tasks (agent-level parallelism) + +How the agent keeps working while a long tool call runs detached. Files: +`navi/core/tasks.py` (TaskManager), `navi/core/task_notes.py` (completion +notes), `navi/core/tool_executor.py` (interception), `navi/tools/tasks.py` +(the `tasks` tool), `navi/api/websocket.py` (message queue). + +The contract follows the pattern used by other harnesses +(start → continue → collect): the agent launches a long call with +`background: true`, immediately receives a `task_id`, keeps doing useful +work, and collects the result either by calling the `tasks` tool or +automatically as a completion note next turn. + +## The `background` parameter + +Tool calls whose arguments carry `"background": true` are detached — the +ToolExecutor submits the whole `tool.execute()` to the TaskManager and returns +a synthetic result right away: + +```json +{"task_id": "bt-ab12cd34", "tool": "terminal", "status": "running", + "args_summary": "{\"action\": \"run\", \"command\": \"make test\"}", + "hint": "The call is running in the background — continue with other work. ..."} +``` + +The `background` flag is stripped before the tool runs. Tools that accept it: +`terminal` (action `run`), `ssh_exec`, `peer` (action `ask`), `spawn_agent`, +`code_exec` — the list is `settings.backgroundable_tools`. Any tool outside +the list runs inline even with the flag set. + +Exception: `terminal` with `action: "open"` keeps its native meaning of +`background` (a persistent detached process) and is never hijacked. + +The interception lives in `ToolExecutor._execute_one`, so it works identically +from the top-level agent loop and from sub-agents. + +## TaskManager + +`navi/core/tasks.py`, process-wide singleton (`get_task_manager()`). Each +`TaskJob` carries: + +- `task_id` — `bt-` + 8 hex chars; +- `status` — `running → completed | failed | cancelled`; +- `result` — the tool's ToolResult; `error` for failures; +- `ring` — a bounded (drop-oldest) event queue with the job's live events; +- `done` — an asyncio event set when the job reaches a terminal state; +- `stop_event` — the job's own stop signal, **independent of the session's**. + +Caps (per session unless noted, configurable in `docs/config.md`): +`tasks_max_per_session=5` running, `tasks_max_global=20`, `tasks_max_spawn=2` +background sub-agents, and a rate limit of `tasks_rate_limit=10` submissions +per 5 minutes. A rejected submit returns the reason as a string; the executor +turns it into a normal failed tool result ("Cannot run in background: …") so +the agent falls back to running the tool inline. + +Finished jobs are reaped by a lazy sweeper after `tasks_ttl_sec` (1 h), with +at most 50 finished jobs kept per session. + +### Isolation guarantees + +- The detached job gets a rebuilt ToolContext and its own ContextVar + overrides: its events stream into its own ring (never into a dead + foreground sink), and `current_stop_event` points at the job's own event — + pressing **Stop on the session does not cancel background tasks**; only + `tasks cancel` does. +- Background jobs never write session history. The only mutation is the + "task started" tool message written by the foreground loop. + +## Delivery of results — two paths + +1. **`task_update` wire event** — published via the orchestrator to all + connected clients the moment the job reaches a terminal state + (`running` on submit, final state on finish). Out-of-band: not part of + the run's replay buffer; on reconnect the client re-syncs via + `tasks list` instead. +2. **Completion note** — recorded in the KV store (`task_notes` scope) and + drained by the next `run_stream` turn into the session context as a + system message (persisted, invisible in the UI): + + ``` + [Background task results] + - bt-ab12 (terminal) completed: PASS 42/42... + ... + (Use these results to continue your work. Do not start new background + tasks in response to this note unless the user asked.) + ``` + + Up to `task_notes_per_turn=5` notes per turn, coalesced; overflow says to + call `tasks check`. At most `task_notes_max_pending=20` notes per session + (oldest dropped). Notes survive a server restart; the running tasks + themselves do not (they die with the process). + +## The `tasks` tool + +`{action: list | check | wait | cancel, task_id?, timeout?}` — session-scoped. + +- `list` — this session's jobs with status and age. +- `check` — status + result (≤2000 chars) + the last ring events as live + progress (useful while a background sub-agent is still working). For + spawn_agent jobs the check output includes the sub-agent's token count. +- `wait` — blocks the turn until the job finishes or `timeout` (capped at + 120 s) expires; `wait_timeout` is a normal failed result telling the agent + to continue other work. +- `cancel` — cooperative cancel: sets the job's stop event and cancels its + asyncio task. + +Manual for the agent: `manuals/tasks.md`. + +## Background sub-agents (spawn_agent) + +`spawn_agent` is synchronous by default (blocks until the sub-agent +completes, ≤300 s). With `background: true` the same interception detaches +it; the sub-agent's event stream (thinking, tool cards, planning) flows into +the job's ring, so `tasks check` shows live progress. Deeper ring (200 +events) for spawn jobs. + +Caveat: tokens consumed by a background sub-agent are reported in +`task_update.subagent_tokens` and the completion note — they no longer +appear in the parent turn's token count. + +## Parallel tool-call batch (Ф3) + +When the profile enables `parallel_tool_calls` (profile override, else +`settings.parallel_tool_calls`, default **off**), a model response carrying +several tool calls executes them concurrently: + +- `tool_started` frames for the whole batch are emitted up-front, in call + order; +- live events from all tools merge through one tagged queue; +- results and tool messages are appended in call order after the batch + settles, with a single session save; +- Stop mid-batch cancels every unfinished tool and synthesises + "stopped by the user" results for each; +- one tool's exception is recorded as its own failed result and does not + kill its neighbours (backend-level LLM errors still abort the turn). + +The gate lives in `Agent._execute_tools_with_sink`; the batch is used only +for 2+ calls. A crash mid-batch can leave assistant tool_calls without their +`role=tool` pair — repaired on session load by +`pg_session_store.repair_dangling_tool_calls` (display history only). + +Agent discipline: batch only **independent** calls; never two actions over +the same terminal in one batch. + +## Message queue + +A user message arriving while a run is active is queued (see +`docs/websocket.md`, `message_queued`) instead of erroring. Queue state is +in-memory on `SessionState` (`pending_user_messages`, maxlen +`message_queue_max=5`, oldest dropped with `dropped_total` counter). Drained +in submit order by the socket after the run finishes, or headlessly when the +client disconnected. \ No newline at end of file diff --git a/docs/websocket.md b/docs/websocket.md index fbd9b8a..a88cd6f 100644 --- a/docs/websocket.md +++ b/docs/websocket.md @@ -89,7 +89,7 @@ The TUI binds this to `/compact` (`Ctrl+X C`). ### Concurrent run guard -Only one agent run may be active per session at a time. If a second message arrives while a run is already in progress (either a WebSocket run or a headless recall), the server rejects it with a WebSocket error. The client should wait for `stream_end` before sending the next message. +Only one agent run may be active per session at a time. If a second message arrives while a run is already in progress (either a WebSocket run or a headless recall), the server **queues** it instead of erroring and replies with a `message_queued` frame (see below). Queued messages execute back-to-back on the same socket once the active run finishes (drained in submit order, each as a normal streaming turn); a max of `message_queue_max` (default 5) are retained — when full, the oldest queued message is dropped silently and `dropped_total` in the frame grows. If the socket disconnects while messages are still queued, the server runs them headlessly (turn-completion push still fires). --- @@ -155,6 +155,8 @@ | `{"type": "mcp_status_update", "server_name": "...", "status": "connected\|disconnected", "tool_count": N, "error": "..."}` | MCP server connection status changed — broadcast to all sessions for toast notifications | | `{"type": "terminal_output", "terminal_name": "...", "stream": "stdout\|stderr", "delta": "..."}` | Streamed stdout/stderr chunk from a background persistent terminal | | `{"type": "terminal_closed", "terminal_name": "...", "reason": "explicit\|idle_timeout\|error\|shutdown\|session_ended"}` | A persistent terminal session ended | +| `{"type": "task_update", "task_id": "bt-...", "session_id": "...", "tool": "...", "status": "running\|completed\|failed\|cancelled", "result_preview": "...", "parent_tool_call_id": "...", "started_at": "...", "finished_at": "...", "subagent_tokens": N}` | A background task (tool call detached with `background: true`, see `docs/tasks.md`) changed state. Out-of-band: NOT part of the run's replay buffer. `parent_tool_call_id` links it to the tool card that spawned it. | +| `{"type": "message_queued", "position": N, "queue_len": N, "max": N, "dropped_total": N}` | The user's message arrived while a run was active and was queued (not an error). It will execute after the current run finishes; `dropped_total` counts messages dropped when the queue was full. | | `{"type": "heartbeat"}` | Periodic keepalive during long silent operations (every 20 s) | | `{"type": "session_sync", "session_id": "...", "profile_id": "..."}` | Client should reload session history from REST (`GET /sessions/{id}`) | diff --git a/manuals/spawn_agent.md b/manuals/spawn_agent.md index 1bbad20..ef9b957 100644 --- a/manuals/spawn_agent.md +++ b/manuals/spawn_agent.md @@ -5,7 +5,7 @@ **One plan step = one spawn_agent call.** If your plan has three AGENT steps, make three separate calls. -**SYNCHRONOUS** — blocks until the sub-agent fully completes or times out (5 minutes hard limit). +**SYNCHRONOUS by default** — blocks until the sub-agent fully completes or times out (5 minutes hard limit). With `"background": true` the call detaches immediately and returns a `task_id` (`bt-...`); the sub-agent keeps running in isolation — see the `tasks` manual for `list`/`check`/`wait`/`cancel`, and PARALLELISM rules in your persona. Background sub-agents are capped separately (`tasks_max_spawn` per session) and their token usage is reported via `task_update`, not in your turn's token count. ## Parameters @@ -16,6 +16,7 @@ | `profile_id` | no | Which profile to use (`secretary`, `server_admin`, `developer`). Defaults to current session's profile. | | `system_prompt` | no | Role specialisation injected into the sub-agent's system prompt between the executor persona and the briefing (e.g. "You are a security auditor. Report findings by severity."). | | `max_iterations` | no | Tool-call iteration limit (default: **40**). | +| `background` | no | `true` → detach immediately, return `task_id`; result arrives later as a completion note (use `tasks` to check/wait/cancel). | ## Sub-agent system prompt structure diff --git a/manuals/tasks.md b/manuals/tasks.md new file mode 100644 index 0000000..a1f9669 --- /dev/null +++ b/manuals/tasks.md @@ -0,0 +1,49 @@ +# tasks — Manual + +## What it does +Manages background tasks you started with `"background": true` on `terminal` (run), `ssh_exec`, `peer`, `spawn_agent` or `code_exec`. A detached call returns a `task_id` like `bt-1a2b3c4d` immediately; the real result arrives later — as a note at the start of your next turn, and via `task_update` events. + +**Background tasks are NOT stopped by run stop.** Only `cancel` kills them. + +## Parameters + +| Parameter | Required | Description | +|-----------|----------|-------------| +| `action` | yes | `list` \| `check` \| `wait` \| `cancel` | +| `task_id` | for `check`/`wait`/`cancel` | The `bt-...` id returned when the task was started. | +| `timeout` | for `wait` | Seconds to block waiting for completion (default 120, hard cap 120). | + +## Actions + +### `list` +All background tasks for this session with status and (for finished ones) a result preview. Use this to re-orient after context loss or reconnect. + +### `check` +Status of one task: `running` / `completed` / `failed` / `cancelled`, plus result preview (≤2000 chars) and, for background sub-agents, the latest events from its progress ring. Non-blocking. + +### `wait` +Blocks the turn until the task finishes or `timeout` elapses. Use ONLY when the next step needs the result — otherwise keep working and let the note arrive. + +### `cancel` +Cooperatively stops a running task (its stop-event fires and its asyncio task is cancelled). Returns an error if the task is not running or belongs to another session. + +## How results reach you + +1. **Completion note** — at the start of your next turn, finished background tasks are injected as a system note `[Background task results]` with one line per task (`bt-... (tool) status: preview`). Up to `task_notes_per_turn` notes are coalesced; older ones stay pending — check them with `list`/`check`. +2. **Live `task_update`** — the user's client shows progress in real time; you don't need to act on it. + +**Do not start new background tasks merely because a results note arrived** — the note is a report, not a request. + +## Usage pattern + +```json +{"action": "list"} +{"action": "check", "task_id": "bt-1a2b3c4d"} +{"action": "wait", "task_id": "bt-1a2b3c4d", "timeout": 30} +{"action": "cancel", "task_id": "bt-1a2b3c4d"} +``` + +## Limits and lifecycle +- Caps: `tasks_max_per_session` (5) concurrent per session, `tasks_max_global` (20) server-wide, `tasks_max_spawn` (2) background sub-agents per session, `tasks_rate_limit` (10 spawns / 5 min per session). Exceeding a cap fails the detach — run the tool inline instead. +- Finished tasks are kept for `tasks_ttl_sec` (1 hour), then reaped; the ring of sub-agent events keeps the last `tasks_event_buffer_size` (50; 200 for spawn tasks). +- Tasks do not survive a server restart. \ No newline at end of file diff --git a/navi/api/websocket.py b/navi/api/websocket.py index 56d3868..df56832 100644 --- a/navi/api/websocket.py +++ b/navi/api/websocket.py @@ -18,12 +18,14 @@ {"type": "tool_started", "tool": "...", "args": {...}, "is_subagent": bool} {"type": "tool_call", "tool": "...", "args": {...}, "result": "...", "success": bool, "is_subagent": bool} {"type": "stream_end", "content": "..."} + {"type": "message_queued", "position": int, "queue_len": int, "max": int, "dropped_total": int} {"type": "context_compressed"} {"type": "error", "message": "..."} """ import asyncio import json +from datetime import datetime, timezone import structlog from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect @@ -104,11 +106,73 @@ user: User | None, cwd: str | None = None, ) -> bool: - """Atomically start an agent run, stream its events, and clean up. + """Start an agent run, stream its events, then drain queued messages. - Returns True if the client stayed connected, False otherwise. When a run - is already in progress for the session, emits an error and returns True - (socket still alive — caller should keep reading). + When a run is already active for the session the message is queued + (``message_queued`` wire event) instead of erroring; once the active run + finishes on this socket, queued messages execute back-to-back. + + Returns True if the client stayed connected, False otherwise. + """ + connected, ran = await _run_single_message( + session_id=session_id, + user_content=user_content, + raw_images=raw_images, + display_content=display_content, + uploaded_files=uploaded_files, + hidden=hidden, + websocket=websocket, + orchestrator=orchestrator, + session_store=session_store, + user=user, + cwd=cwd, + ) + # Messages queued while the run(s) above were active — execute them in + # order on the same socket. Entries carry the submitting user so each + # drained run runs with its original identity. Only drain after a run + # actually executed here: when this call itself got queued on a busy + # session, the run that owns the socket will drain the queue. + while connected and ran: + entry = orchestrator.pop_pending(session_id) + if entry is None: + break + log.info("ws.queued_drain", session_id=session_id) + connected, ran = await _run_single_message( + session_id=session_id, + user_content=entry["user_content"], + raw_images=entry.get("raw_images"), + display_content=entry.get("display_content"), + uploaded_files=entry.get("uploaded_files") or [], + hidden=entry.get("hidden", False), + websocket=websocket, + orchestrator=orchestrator, + session_store=session_store, + user=entry.get("user"), + cwd=entry.get("cwd"), + ) + return connected + + +async def _run_single_message( + *, + session_id: str, + user_content: str, + raw_images: list[str] | None, + display_content: str | None, + uploaded_files: list[dict], + hidden: bool, + websocket: WebSocket, + orchestrator: "AgentSessionOrchestrator", + session_store: SessionStore, + user: User | None, + cwd: str | None = None, +) -> tuple[bool, bool]: + """Run one user message through the agent and stream it to the client. + + Returns ``(connected, ran)``: ``ran`` is False when the message was + queued on a busy session instead of executing. When a run is already in + progress the message is queued and ``message_queued`` is emitted (socket + stays alive — caller keeps reading). """ run = None queue: asyncio.Queue | None = None @@ -116,11 +180,26 @@ # Guard against concurrent runs for the same session (atomically). async with orchestrator.session_lock(session_id): if orchestrator.is_running(session_id): + entry = { + "user_content": user_content, + "raw_images": raw_images, + "display_content": display_content, + "uploaded_files": uploaded_files, + "hidden": hidden, + "user": user, + "cwd": cwd, + "queued_at": datetime.now(timezone.utc).isoformat(), + } + qinfo = orchestrator.queue_message(session_id, entry) + log.info("ws.message_queued", session_id=session_id, **qinfo) await websocket.send_json({ - "type": "error", - "message": "Agent is already running for this session.", + "type": "message_queued", + "position": qinfo["position"], + "queue_len": qinfo["queue_len"], + "max": settings.message_queue_max, + "dropped_total": qinfo["dropped_total"], }) - return True # socket still alive; caller should keep reading + return (True, False) # socket alive; the active run will drain # Register run and subscribe before starting the task so we never # miss events even if the task is very fast. @@ -164,7 +243,7 @@ await websocket.send_json({"type": "stream_start"}) connected = await _stream_to_client(websocket, queue) - return connected + return (connected, True) finally: if queue is not None and run is not None: run.unsubscribe(queue) diff --git a/navi/config.py b/navi/config.py index e6280d8..c33c90c 100644 --- a/navi/config.py +++ b/navi/config.py @@ -75,6 +75,22 @@ ws_replay_buffer_size: int = 500 # max events retained for WS reconnect replay share_file_max_size_mb: int = 1024 + # Background tasks (see docs/tasks.md). Tools in `backgroundable_tools` + # accept background=true: the call is detached into a TaskManager job and + # the tool call returns a task_id immediately. Tasks are in-memory — a + # restart loses them (same as terminal processes). + tasks_max_per_session: int = 5 # concurrent running tasks per session + tasks_max_global: int = 20 # concurrent running tasks server-wide + tasks_max_spawn: int = 2 # concurrent background spawn_agent per session + tasks_ttl_sec: int = 3600 # finished-task retention before reap + tasks_event_buffer_size: int = 50 # per-task event ring (spawn tasks: 200) + tasks_rate_limit: int = 10 # max task spawns per 5 min per session + task_notes_max_pending: int = 20 # pending completion notes per session + task_notes_per_turn: int = 5 # max notes coalesced into one turn injection + message_queue_max: int = 5 # user messages queued while a run is active + backgroundable_tools: str = "terminal,ssh_exec,peer,spawn_agent,code_exec" + parallel_tool_calls: bool = False # execute a multi-tool-call batch concurrently + # Public base URL used by share_file tool to build download links. # Change if the server is behind a reverse proxy or runs on a different port. public_url: str = "http://localhost:8099" diff --git a/navi/core/agent.py b/navi/core/agent.py index a680ccc..42f6168 100644 --- a/navi/core/agent.py +++ b/navi/core/agent.py @@ -87,6 +87,24 @@ _TOOL_DONE = object() +class _TaggedSink: + """Routes one tool's live events into the shared batch queue, tagged by index. + + Mirrors the asyncio.Queue surface the sinks actually use (``put`` / + ``put_nowait``) so it is a drop-in replacement for a per-tool queue. + """ + + def __init__(self, queue: asyncio.Queue, idx: int) -> None: + self._queue = queue + self._idx = idx + + async def put(self, item) -> None: + await self._queue.put((self._idx, item)) + + def put_nowait(self, item) -> None: + self._queue.put_nowait((self._idx, item)) + + async def _todo_progress_message( session_id: str, *, first_iteration: bool = False ) -> "Message | None": @@ -180,6 +198,30 @@ full_content = event.full_content or "" return full_content + async def _drain_task_notes(self, session_id: str, session) -> None: + """Move finished-background-task notes into the session context. + + Called once at the start of run_stream(), before the first LLM call: + the note becomes part of the LLM-visible context (and the compression + window) but is never displayed — clients learned about completions via + the live task_update event. Only the foreground run_stream mutates the + session, so there is no race with background jobs. + """ + from navi.core import task_notes + + try: + note = await task_notes.drain(session_id) + except Exception: + log.exception("agent.task_notes_drain_failed", session_id=session_id) + return + if not note: + return + session.context.append( + Message(role="system", content=note, metadata={"source": "task_note"}) + ) + await self._sessions.save(session) + log.info("agent.task_notes_drained", session_id=session_id) + async def run_ephemeral( self, user_message: str, @@ -425,6 +467,11 @@ else: log.debug("agent.memory_facts_none", session_id=session_id) + # Drain background-task completion notes into the session context + # (persisted so compression keeps them; not displayed — the client + # already saw the live task_update event). + await self._drain_task_notes(session_id, session) + anti_stall = AntiStallMonitor(profile) if profile.anti_stall_enabled: await anti_stall.init(session_id) @@ -627,6 +674,14 @@ user_info=current_user_info.get(), cwd=_cwd_var.get(), ) + # Parallel batch gate: profile override wins, else the global flag. + if turn_ctx.parallel_tool_calls is False: + if profile.parallel_tool_calls is None: + from navi.config import settings as _settings + + turn_ctx.parallel_tool_calls = _settings.parallel_tool_calls + else: + turn_ctx.parallel_tool_calls = profile.parallel_tool_calls # Expose a per-iteration PlanRunner so the `plan` tool can run the # planner over the live session. Constructed inside the loop (not # once before) using this iteration's profile/llm/tool_schemas so a @@ -879,7 +934,165 @@ async def _execute_tools_with_sink( self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None ): - """Execute tool calls with cooperative stop support. + """Execute tool calls, sequentially or as a parallel batch. + + The mode is decided per-turn by ``turn_ctx.parallel_tool_calls`` + (profile override ? global setting, resolved at tool_ctx build). + """ + if turn_tool_calls and turn_ctx.parallel_tool_calls and len(turn_tool_calls) > 1: + async for ev in self._execute_tools_parallel( + turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx + ): + yield ev + else: + async for ev in self._execute_tools_sequential( + turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx + ): + yield ev + + async def _execute_tools_parallel( + self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None + ): + """Execute an independent batch of tool calls concurrently. + + ToolStarted is emitted for every call up-front (in call order) so the + UI shows the whole batch immediately. Live events from all tools merge + through one tagged queue; results and tool messages are appended in + call order after the batch settles, with a single session save. + + Only batch independent calls — two actions over the same terminal in + one batch race each other. + """ + from navi.tools.todo import started_metadata_for_call + + tool_map = {t.name: t for t in tools} + for tc in turn_tool_calls: + yield ToolStarted( + tool_name=tc.name, + arguments=tc.arguments, + tool_call_id=tc.id, + metadata=await started_metadata_for_call(tc, tool_ctx), + ) + + shared: asyncio.Queue = asyncio.Queue() + holders: list[list] = [[] for _ in turn_tool_calls] + tasks: list[asyncio.Task] = [] + + for idx, tc in enumerate(turn_tool_calls): + tagged = _TaggedSink(shared, idx) + sink_token = current_event_sink.set(tagged) + + async def _run_with_sentinel(_tc=tc, _idx=idx, _holder=holders[idx], _sink=tagged): + try: + _holder.append( + await self._tool_executor._run_single_tool(_tc, tool_map, ctx=tool_ctx) + ) + except Exception as exc: + _holder.append(exc) + finally: + await _sink.put(_TOOL_DONE) + + tasks.append(asyncio.create_task(_run_with_sentinel())) + current_event_sink.reset(sink_token) + + try: + pending = set(range(len(tasks))) + stopped = False + while pending: + try: + item = await asyncio.wait_for(shared.get(), timeout=1.0) + except asyncio.TimeoutError: + if stop_event and stop_event.is_set(): + stopped = True + break + continue + + _idx, value = item + if value is _TOOL_DONE: + pending.discard(_idx) + continue + if isinstance(value, SubagentComplete): + turn_ctx.subagent_tokens += value.token_count + turn_ctx.tool_call_count += value.tool_call_count + elif isinstance(value, AIHelperTokensUsed): + turn_ctx.subagent_tokens += value.completion_tokens + else: + yield value + + if stopped: + log.info( + "agent.batch_stopped", + tools=[tc.name for tc in turn_tool_calls], + ) + for tc in turn_tool_calls: + yield ToolEvent( + tool_name=tc.name, + arguments=tc.arguments, + result="Tool execution was stopped by the user.", + success=False, + tool_call_id=tc.id, + ) + session.messages.append( + Message( + role="tool", + content="Tool execution was stopped by the user.", + tool_call_id=tc.id, + name=tc.name, + metadata={}, + is_context=False, + ) + ) + await self._sessions.save(session) + return + + # Results and messages in call order, one save for the batch. + for tc, holder in zip(turn_tool_calls, holders): + r = ( + holder[0] + if holder + else RuntimeError("tool task produced no result") + ) + if isinstance(r, Exception): + if isinstance(r, (LLMBackendError, LLMConnectionError)): + raise r + log.warning("agent.tool_exception", tool=tc.name, error=str(r)) + tool_event = ToolEvent( + tool_name=tc.name, + arguments=tc.arguments, + result=f"Error: {r}", + success=False, + tool_call_id=tc.id, + ) + msg = Message( + role="tool", + content=f"Error: {r}", + tool_call_id=tc.id, + name=tc.name, + metadata={}, + ) + image_msg = None + else: + tool_event, msg, image_msg = r + + turn_ctx.tool_call_count += 1 + yield tool_event + session.messages.append(msg) + session.context.append(msg) + if image_msg: + session.messages.append(image_msg) + session.context.append(image_msg) + await self._sessions.save(session) + finally: + # gather(return_exceptions=True) contains each cancelled tool task + # (awaiting one directly would re-raise CancelledError past the + # results already produced) while an outer cancellation of the run + # itself still propagates. + await asyncio.gather(*tasks, return_exceptions=True) + + async def _execute_tools_sequential( + self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None + ): + """Execute tool calls one at a time with cooperative stop support. Polls *stop_event* every second while draining the event sink so the Stop button works even during long-running tools (terminal, SSH, @@ -1001,7 +1214,8 @@ finally: if not tool_task.done(): tool_task.cancel() - try: - await tool_task - except Exception: - pass + # gather(return_exceptions=True) contains the cancelled tool + # task (awaiting it directly would re-raise CancelledError and + # abort the turn before the synthetic stopped results land), + # while an outer cancellation of the run itself propagates. + await asyncio.gather(tool_task, return_exceptions=True) diff --git a/navi/core/agent_run_context.py b/navi/core/agent_run_context.py index f290205..e60d020 100644 --- a/navi/core/agent_run_context.py +++ b/navi/core/agent_run_context.py @@ -34,6 +34,9 @@ # still had open steps. Bounded by profile.final_intercept_limit so a model # that keeps producing bare text is eventually allowed to finalise. final_interceptions: int = 0 + # Parallel tool-call batch enabled for this turn (profile override ? global + # setting). Resolved once at tool_ctx build; False keeps the sequential path. + parallel_tool_calls: bool = False @dataclass diff --git a/navi/core/container.py b/navi/core/container.py index e58040b..505b72e 100644 --- a/navi/core/container.py +++ b/navi/core/container.py @@ -232,4 +232,23 @@ # WebSocket(s) (terminal_opened/output/closed) — the manager emits them, # the orchestrator fans them out per session_id. terminal_manager.set_event_callback(container.orchestrator._on_terminal_event) + + # Background tasks: publish TaskUpdate through the same out-of-band session + # notify path, persist completion notes in the shared KV store, and keep + # the TTL sweeper running. + from navi.core.tasks import get_task_manager + from navi.core import task_notes + + def _on_task_update(update) -> None: + import asyncio + + try: + asyncio.get_running_loop().create_task( + container.orchestrator.notify_task_update(update) + ) + except RuntimeError: + logger.warning("task_update without running loop dropped: %s", update.task_id) + + get_task_manager().set_update_callback(_on_task_update) + task_notes.set_kv_store(kv_store) return container diff --git a/navi/core/events.py b/navi/core/events.py index be12122..5f75e85 100644 --- a/navi/core/events.py +++ b/navi/core/events.py @@ -404,9 +404,44 @@ } +@dataclass +class TaskUpdate: + """Emitted when a background task changes state (started/finished/cancelled). + + Delivered out-of-band via the orchestrator's session-notify path (like + terminal events) — NOT through a run's event sink or replay buffer, because + background tasks outlive any single run. + """ + + task_id: str + session_id: str + tool: str + status: str # "running" | "completed" | "failed" | "cancelled" + result_preview: str = "" + parent_tool_call_id: str = "" # binds subagent progress to its spawn card + started_at: str | None = None # ISO timestamps (JSON-friendly) + finished_at: str | None = None + subagent_tokens: int | None = None + + def to_wire(self) -> dict: + return { + "type": "task_update", + "task_id": self.task_id, + "session_id": self.session_id, + "tool": self.tool, + "status": self.status, + "result_preview": self.result_preview, + "parent_tool_call_id": self.parent_tool_call_id, + "started_at": self.started_at, + "finished_at": self.finished_at, + "subagent_tokens": self.subagent_tokens, + } + + AgentEvent = ( ToolStarted | ToolEvent | TextDelta | ThinkingDelta | ThinkingEnd | StreamEnd | StreamStopped | CompressionStarted | ContextCompressed | TurnThinking | ProfileSwitched | PlanningStatus | PlanReady | SubagentComplete | AIHelperTokensUsed | PlanningDebugData | RecallUpdate | McpStatusUpdate | TerminalOutputDelta | TerminalOpened | TerminalClosed | TodoUpdated + | TaskUpdate ) diff --git a/navi/core/orchestrator.py b/navi/core/orchestrator.py index a3a793c..9f43981 100644 --- a/navi/core/orchestrator.py +++ b/navi/core/orchestrator.py @@ -4,6 +4,7 @@ import asyncio import dataclasses +from collections import deque from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any @@ -60,6 +61,10 @@ busy_event: asyncio.Event | None = None websockets: list[Any] = field(default_factory=list) terminals: dict[str, Any] = field(default_factory=dict) + # User messages that arrived while a run was active — replayed as normal + # turns after the current one finishes (oldest dropped silently when full). + pending_user_messages: deque = field(default_factory=lambda: deque(maxlen=5)) + dropped_queued_messages: int = 0 def _event_to_dict(event) -> dict | None: @@ -75,6 +80,12 @@ self._container = container self._sessions: dict[str, SessionState] = {} self._session_locks: dict[str, asyncio.Lock] = {} + # Strong refs for fire-and-forget queued-message drain tasks so they + # are not garbage-collected mid-run. + self._drain_tasks: set[asyncio.Task] = set() + # Sessions currently being drained headlessly (guards against the + # drain's own run_agent finally re-scheduling a parallel drain). + self._headless_draining: set[str] = set() # Wire event bus subscriber so recall updates reach connected clients from navi.core.event_bus import get_event_bus @@ -130,6 +141,102 @@ if payload: await self._notify_session(session_id, payload) + async def notify_task_update(self, update: Any) -> None: + """Deliver a background TaskUpdate to that session's clients. + + Out-of-band like terminal events — a background task outlives any + single run, so it bypasses the run's event sink and replay buffer. + """ + from navi.core.events import TaskUpdate + + if isinstance(update, TaskUpdate): + await self._notify_session(update.session_id, update.to_wire()) + + # -- user message queue (messages arriving while a run is active) ------- + + def queue_message(self, session_id: str, entry: dict) -> dict: + """Enqueue a user message that arrived during an active run. + + Returns {position, queue_len, dropped_total}. The deque's maxlen drops + the oldest silently — tracked in dropped_queued_messages so the client + can be warned. + """ + state = self._get_or_create_state(session_id) + if len(state.pending_user_messages) == state.pending_user_messages.maxlen: + state.dropped_queued_messages += 1 + state.pending_user_messages.append(entry) + return { + "position": len(state.pending_user_messages), + "queue_len": len(state.pending_user_messages), + "dropped_total": state.dropped_queued_messages, + } + + def has_pending(self, session_id: str) -> bool: + state = self._sessions.get(session_id) + return bool(state and state.pending_user_messages) + + def pop_pending(self, session_id: str) -> dict | None: + state = self._sessions.get(session_id) + if state and state.pending_user_messages: + return state.pending_user_messages.popleft() + return None + + # ── headless drain of queued messages (client disconnected) ───────────── + + def schedule_queued_drain(self, session_id: str, session_store) -> None: + """After a run finishes: execute leftover queued messages headlessly. + + Only when no client tab is watching the session — a live socket drains + the queue itself in ``_start_agent_run``. Fire-and-forget; the entry + ordering and user identity are preserved from the queued entries. + """ + if not self.has_pending(session_id): + return + if session_id in self._headless_draining: + return + state = self._sessions.get(session_id) + if state is not None and state.websockets: + return + task = asyncio.create_task(self._drain_queued(session_id, session_store)) + self._drain_tasks.add(task) + task.add_done_callback(self._drain_tasks.discard) + + async def _drain_queued(self, session_id: str, session_store) -> None: + """Run queued user messages without a websocket (push still fires).""" + self._headless_draining.add(session_id) + try: + await self._drain_queued_inner(session_id, session_store) + finally: + self._headless_draining.discard(session_id) + + async def _drain_queued_inner(self, session_id: str, session_store) -> None: + """Run queued user messages without a websocket (push still fires).""" + while True: + entry = self.pop_pending(session_id) + if entry is None: + return + async with self.session_lock(session_id): + if self.is_running(session_id): + # Another run (websocket or recall) took over — it will + # drain the rest of the queue itself. + self.queue_message(session_id, entry) + return + self.create_run(session_id) + try: + await self.run_agent( + session_id, + entry["user_content"], + entry.get("raw_images"), + entry.get("display_content"), + entry.get("uploaded_files") or [], + session_store, + cwd=entry.get("cwd"), + hidden=entry.get("hidden", False), + ) + except Exception: + log.exception("orchestrator.queued_drain_failed", session_id=session_id) + return + def _get_or_create_state(self, session_id: str) -> SessionState: state = self._sessions.get(session_id) if state is None: @@ -148,6 +255,7 @@ and state.busy_event is None and not state.websockets and not state.terminals + and not state.pending_user_messages # queued drain still reads them ): self._sessions.pop(session_id, None) self._session_locks.pop(session_id, None) @@ -326,6 +434,9 @@ if state is not None: state.run = None await self._cleanup(session_id) + # Client may have disconnected with messages still queued — run + # them headlessly (a live socket would have drained them itself). + self.schedule_queued_drain(session_id, session_store) # ── Forced compact (TUI /compact) ─────────────────────────────────────── diff --git a/navi/core/pg_session_store.py b/navi/core/pg_session_store.py index 9df770c..456eb34 100644 --- a/navi/core/pg_session_store.py +++ b/navi/core/pg_session_store.py @@ -136,6 +136,47 @@ return [Message.model_validate(m) for m in json.loads(raw)] +def repair_dangling_tool_calls(messages: list[Message]) -> int: + """Synthesize placeholder results for assistant tool_calls that lost their + role=tool pair (a crash mid-batch, e.g. a parallel batch killed by a server + restart). Returns the number of placeholders added. + + Placeholders are inserted right after the last tool message of the affected + batch (before the next non-tool message) and marked ``is_context=False`` — + display history stays coherent, the LLM context is untouched. + """ + answered = { + m.tool_call_id for m in messages if m.role == "tool" and m.tool_call_id + } + out: list[Message] = [] + pending: list[Message] = [] + for m in messages: + if m.role != "tool" and pending: + out.extend(pending) + pending = [] + out.append(m) + if m.role == "assistant" and m.tool_calls: + for tc in m.tool_calls: + if not tc.id or tc.id in answered: + continue + pending.append( + Message( + role="tool", + content="[Interrupted before result]", + tool_call_id=tc.id, + name=tc.name, + metadata={}, + is_context=False, + ) + ) + out.extend(pending) + if len(out) != len(messages): + added = len(out) - len(messages) + messages[:] = out + return added + return 0 + + def _message_key(m: Message) -> tuple: """Stable key for matching a message between messages[] and context[]. @@ -366,6 +407,9 @@ archive_threshold, ) all_messages = [_row_to_message(r) for r in all_rows] + # A crash mid-tool-batch (e.g. parallel batch) can leave assistant + # tool_calls without their role=tool pair — patch display history. + repair_dangling_tool_calls(all_messages) messages = [m for m in all_messages if m.is_display] context = [m for m in all_messages if m.is_context] diff --git a/navi/core/registry.py b/navi/core/registry.py index 865de4e..a8a774e 100644 --- a/navi/core/registry.py +++ b/navi/core/registry.py @@ -24,6 +24,7 @@ SshExecTool, ScratchpadTool, SwitchProfileTool, + TasksTool, TerminalTool, TestMcpToolTool, TodoTool, @@ -246,7 +247,8 @@ reload_tool, list_tool, manual_tool, mcp_status_tool, create_mcp_server_tool, test_mcp_tool_tool, schedule_recall_tool, manage_recall_tool, - spawn_tool, switch_tool, list_profiles_tool] + spawn_tool, switch_tool, list_profiles_tool, + TasksTool()] if memory_tool: builtins.append(memory_tool) for builtin in builtins: diff --git a/navi/core/task_notes.py b/navi/core/task_notes.py new file mode 100644 index 0000000..b873cc1 --- /dev/null +++ b/navi/core/task_notes.py @@ -0,0 +1,110 @@ +"""Pending completion notes for background tasks. + +When a background task finishes, the TaskManager records a note here. The +next ``run_stream`` turn drains the notes and injects them into the session +context as a system message (persisted, ``is_display=False``) — the agent +learns the result without the client re-rendering anything (it already saw +the live ``task_update`` event). + +Notes are persisted in the shared KV store (scope ``task_notes``) so they +survive a server restart even though the tasks themselves do not. +""" + +import asyncio +import json + +import structlog + +log = structlog.get_logger() + +_kv_store = None + + +def set_kv_store(kv) -> None: + """Inject the shared KvStore instance (called once at startup).""" + global _kv_store + _kv_store = kv + + +async def add_note(job) -> None: + """Record a completion note for a finished TaskJob.""" + if _kv_store is None: + log.warning("tasks.notes_no_store", task_id=job.task_id) + return + from navi.config import settings + + async with _lock(): + notes = await _load(job.session_id) + notes.append({ + "task_id": job.task_id, + "tool": job.tool, + "status": job.status, + "preview": job.preview(limit=800), + "subagent_tokens": job.subagent_tokens, + }) + max_pending = settings.task_notes_max_pending + dropped = 0 + if len(notes) > max_pending: + dropped = len(notes) - max_pending + notes = notes[-max_pending:] + await _save(job.session_id, notes, dropped) + + +async def drain(session_id: str) -> str | None: + """Take up to task_notes_per_turn notes and coalesce into one text block.""" + if _kv_store is None: + return None + from navi.config import settings + + async with _lock(): + notes = await _load(session_id) + if not notes: + return None + take = notes[: settings.task_notes_per_turn] + rest = notes[len(take):] + await _save(session_id, rest, 0) + + lines = [f"- {n['task_id']} ({n['tool']}) {n['status']}: {n['preview'] or '(no output)'}" + for n in take] + text = "[Background task results]\n" + "\n".join(lines) + if rest: + text += ( + f"\n... {len(rest)} older result(s) not shown — " + f"call tasks check to inspect them." + ) + text += ( + "\n(Use these results to continue your work. Do not start new " + "background tasks in response to this note unless the user asked.)" + ) + return text + + +_lock_instance = asyncio.Lock() + + +def _lock() -> asyncio.Lock: + return _lock_instance + + +async def _load(session_id: str) -> list[dict]: + raw = await _kv_store.get("", session_id, "task_notes", "pending") + if not raw: + return [] + try: + data = json.loads(raw) + return data if isinstance(data, list) else [] + except (ValueError, TypeError): + return [] + + +async def _save(session_id: str, notes: list[dict], dropped: int) -> None: + payload = json.dumps(notes, ensure_ascii=False) + if dropped: + log.warning("tasks.notes_dropped", session_id=session_id, dropped=dropped) + await _kv_store.set("", session_id, "task_notes", "pending", payload) + + +async def pending_count(session_id: str) -> int: + if _kv_store is None: + return 0 + return len(await _load(session_id)) \ No newline at end of file diff --git a/navi/core/tasks.py b/navi/core/tasks.py new file mode 100644 index 0000000..759b11a --- /dev/null +++ b/navi/core/tasks.py @@ -0,0 +1,408 @@ +"""Background task manager — runs detached tool executions for the agent. + +A tool call with ``background: true`` (only for tools listed in +``settings.backgroundable_tools``) is submitted here by the ToolExecutor: the +tool's coroutine keeps running in its own asyncio task while the agent turn +continues. The caller immediately gets a ToolResult carrying the task_id. + +Lifecycle: running → completed | failed | cancelled. On terminal state the +manager publishes a ``TaskUpdate`` event (out-of-band, via a callback wired to +the orchestrator) and records a pending note that the next turn's +``run_stream`` drains into the session context (see navi/core/task_notes.py). + +Tasks are in-memory: a server restart loses them (terminal processes die with +the server anyway). Session Stop does NOT cancel background tasks — only +``tasks cancel`` does; each job carries its own stop_event, independent from +the run's. +""" + +import asyncio +import contextvars +import json +import time +import uuid +from collections import deque +from dataclasses import dataclass, field +from datetime import UTC, datetime + +import structlog + +from navi.tools._internal.base import ( + ToolContext, + ToolResult, + current_event_sink, + current_stop_event, +) +from navi.core.events import SubagentComplete, TaskUpdate + +log = structlog.get_logger() + + +class BoundedEventQueue(asyncio.Queue): + """Non-blocking ring queue for background-task events. + + ``put`` never waits: when full, the oldest event is dropped. Producers + (subagent_runner does ``await sink.put(...)``) must never block or raise — + a detached task with a dead consumer must not leak or stall. + """ + + def __init__(self, maxsize: int = 50) -> None: + super().__init__(maxsize=maxsize) + + async def put(self, item) -> None: # noqa: D401 — asyncio.Queue signature + while self.full(): + try: + self.get_nowait() + except asyncio.QueueEmpty: + break + self.put_nowait(item) + + def put_nowait(self, item) -> None: + while self.full(): + try: + self.get_nowait() + except asyncio.QueueEmpty: + break + super().put_nowait(item) + + +@dataclass +class TaskJob: + task_id: str + session_id: str + tool: str + args: dict # background flag stripped + parent_tool_call_id: str = "" + status: str = "running" # running | completed | failed | cancelled + result: ToolResult | None = None + error: str | None = None + created_at: float = field(default_factory=time.time) + finished_at: float | None = None + ring: BoundedEventQueue | None = None + done: asyncio.Event = field(default_factory=asyncio.Event) + stop_event: asyncio.Event = field(default_factory=asyncio.Event) + task: asyncio.Task | None = None + subagent_tokens: int | None = None + + def preview(self, limit: int = 800) -> str: + """Short result summary for notes / task_update events.""" + if self.status == "running": + return "" + if self.result is None: + return self.error or self.status + text = self.result.to_message_content() + return text[:limit] + + def to_update(self) -> TaskUpdate: + return TaskUpdate( + task_id=self.task_id, + session_id=self.session_id, + tool=self.tool, + status=self.status, + result_preview=self.preview(), + parent_tool_call_id=self.parent_tool_call_id, + started_at=datetime.fromtimestamp(self.created_at, tz=UTC).isoformat(), + finished_at=( + datetime.fromtimestamp(self.finished_at, tz=UTC).isoformat() + if self.finished_at is not None + else None + ), + subagent_tokens=self.subagent_tokens, + ) + + +class TaskManager: + """Registry of detached tool executions.""" + + FINISHED_CAP_PER_SESSION = 50 + + def __init__(self) -> None: + self._jobs: dict[str, TaskJob] = {} + self._submit_times: dict[str, deque[float]] = {} # session_id → timestamps + self._update_callback = None # callable(TaskUpdate) | None + self._sweeper_task: asyncio.Task | None = None + + # -- wiring ------------------------------------------------------------- + + def set_update_callback(self, callback) -> None: + """Inject the orchestrator notifier (called once at container build).""" + self._update_callback = callback + + def _publish(self, job: TaskJob) -> None: + if self._update_callback is None: + return + try: + self._update_callback(job.to_update()) + except Exception: + log.exception("tasks.publish_failed", task_id=job.task_id) + + # -- caps --------------------------------------------------------------- + + def _rate_limited(self, session_id: str) -> bool: + times = self._submit_times.setdefault(session_id, deque()) + now = time.time() + while times and now - times[0] > 300: + times.popleft() + from navi.config import settings + + return len(times) >= settings.tasks_rate_limit + + def _cap_reason(self, session_id: str, tool: str) -> str | None: + from navi.config import settings + + running = [j for j in self._jobs.values() if j.status == "running"] + if len(running) >= settings.tasks_max_global: + return "global task limit reached" + per_session = [j for j in running if j.session_id == session_id] + if len(per_session) >= settings.tasks_max_per_session: + return "per-session task limit reached" + if tool == "spawn_agent": + spawns = [j for j in per_session if j.tool == "spawn_agent"] + if len(spawns) >= settings.tasks_max_spawn: + return "background subagent limit reached" + return None + + # -- API ---------------------------------------------------------------- + + def submit( + self, + session_id: str, + tool: str, + args: dict, + coro_factory, + ctx: ToolContext | None, + parent_tool_call_id: str = "", + ring_size: int | None = None, + ) -> TaskJob | str: + """Detach a tool execution. Returns a TaskJob, or a rejection reason. + + ``coro_factory`` is a one-arg callable receiving the rebuilt background + ToolContext and returning the coroutine to run (invoked inside the + detached task, after ContextVar overrides). Never close over the + caller's own ToolContext. + """ + if self._rate_limited(session_id): + return "task rate limit reached (try again later)" + reason = self._cap_reason(session_id, tool) + if reason: + return reason + + from navi.config import settings + + job = TaskJob( + task_id=f"bt-{uuid.uuid4().hex[:8]}", + session_id=session_id, + tool=tool, + args=args, + parent_tool_call_id=parent_tool_call_id, + ring=BoundedEventQueue(maxsize=ring_size or settings.tasks_event_buffer_size), + ) + job.task = asyncio.create_task( + self._run_job(job, coro_factory, ctx), + name=f"bgtask-{job.task_id}", + ) + self._jobs[job.task_id] = job + self._submit_times.setdefault(session_id, deque()).append(time.time()) + self._ensure_sweeper() + self._publish(job) + log.info( + "tasks.submitted", task_id=job.task_id, tool=tool, + session_id=session_id, parent_tool_call_id=parent_tool_call_id, + ) + return job + + def get(self, task_id: str, session_id: str) -> TaskJob | None: + job = self._jobs.get(task_id) + if job is not None and job.session_id == session_id: + return job + return None + + def list(self, session_id: str) -> list[TaskJob]: + return sorted( + (j for j in self._jobs.values() if j.session_id == session_id), + key=lambda j: j.created_at, + ) + + def cancel(self, job: TaskJob) -> bool: + if job.status != "running" or job.task is None or job.task.done(): + return False + job.stop_event.set() + job.task.cancel() + # A task cancelled before its first step never enters _run_job, so its + # CancelledError handler never fires — finalise by hand via the + # done-callback (a no-op when _run_job already finalised). + job.task.add_done_callback(lambda _t: self._finalise_if_unfinished(job)) + return True + + def _finalise_if_unfinished(self, job: TaskJob) -> None: + if job.status != "running": + return # normal path — _run_job already finalised + job.status = "cancelled" + job.error = "Task was cancelled." + job.result = ToolResult(success=False, output="Task was cancelled.", + error="cancelled") + job.finished_at = time.time() + self._harvest_subagent_tokens(job) + self._publish(job) + job.done.set() + log.info("tasks.finished", task_id=job.task_id, tool=job.tool, + status="cancelled", session_id=job.session_id) + + async def _note(): + try: + from navi.core.task_notes import add_note + + await add_note(job) + except Exception: + log.exception("tasks.note_write_failed", task_id=job.task_id) + + asyncio.create_task(_note()) + + def running_count(self, session_id: str) -> int: + return sum( + 1 for j in self._jobs.values() + if j.session_id == session_id and j.status == "running" + ) + + # -- execution ---------------------------------------------------------- + + async def _run_job(self, job: TaskJob, coro_factory, ctx: ToolContext | None) -> None: + """Run the detached coroutine with per-task sink/stop overrides.""" + from navi.tools._internal.base import ( + current_user_id, + current_user_role, + current_user_info, + current_session_id, + ) + + # Copy the caller's context, then override the run-scoped vars so the + # detached job (a) streams events into its own bounded ring instead of + # the dead foreground sink, and (b) ignores the session's Stop signal. + bg_ctx = ToolContext( + session_id=getattr(ctx, "session_id", None), + event_sink=job.ring, + stop_event=job.stop_event, + model=getattr(ctx, "model", None), + user_id=getattr(ctx, "user_id", None), + user_role=getattr(ctx, "user_role", "user"), + user_info=getattr(ctx, "user_info", None), + cwd=getattr(ctx, "cwd", None), + ) + overrides: list[tuple[contextvars.ContextVar, object]] = [ + (current_event_sink, job.ring), + (current_stop_event, job.stop_event), + (current_session_id, job.session_id), + ] + if ctx is not None: + overrides += [ + (current_user_id, getattr(ctx, "user_id", None)), + (current_user_role, getattr(ctx, "user_role", "user")), + (current_user_info, getattr(ctx, "user_info", None)), + ] + + for var, value in overrides: + var.set(value) + try: + job.result = await coro_factory(bg_ctx) + job.status = "completed" + except asyncio.CancelledError: + job.status = "cancelled" + job.error = "Task was cancelled." + job.result = ToolResult(success=False, output="Task was cancelled.", + error="cancelled") + except Exception as e: + job.status = "failed" + job.error = f"{type(e).__name__}: {e}" + job.result = ToolResult(success=False, output=str(e), error=type(e).__name__) + finally: + job.finished_at = time.time() + # ContextVar reset is unnecessary: the task's context dies with it. + self._harvest_subagent_tokens(job) + self._publish(job) + job.done.set() # never gated on the note write below + try: + from navi.core.task_notes import add_note + + await add_note(job) + except Exception: + log.exception("tasks.note_write_failed", task_id=job.task_id) + log.info( + "tasks.finished", task_id=job.task_id, tool=job.tool, + status=job.status, session_id=job.session_id, + ) + + def _harvest_subagent_tokens(self, job: TaskJob) -> None: + """Pull SubagentComplete token counts out of the drained ring.""" + if job.ring is None: + return + tokens = None + while True: + try: + item = job.ring.get_nowait() + except asyncio.QueueEmpty: + break + if isinstance(item, SubagentComplete): + tokens = item.token_count + job.subagent_tokens = tokens + + # -- maintenance -------------------------------------------------------- + + def _ensure_sweeper(self) -> None: + if self._sweeper_task is None or self._sweeper_task.done(): + self._sweeper_task = asyncio.create_task(self._sweep_loop()) + + async def _sweep_loop(self) -> None: + from navi.config import settings + + while True: + await asyncio.sleep(60) + self.reap() + + def reap(self) -> int: + """Drop finished jobs past TTL / cap. Returns number reaped.""" + from navi.config import settings + + now = time.time() + reaped = 0 + finished_per_session: dict[str, int] = {} + for job in list(self._jobs.values()): + if job.status == "running": + continue + finished_per_session[job.session_id] = ( + finished_per_session.get(job.session_id, 0) + 1 + ) + age = now - (job.finished_at or now) + if age > settings.tasks_ttl_sec: + del self._jobs[job.task_id] + reaped += 1 + # Enforce the finished-jobs cap (oldest first) per session. + for session_id, count in finished_per_session.items(): + if count <= self.FINISHED_CAP_PER_SESSION: + continue + finished = sorted( + (j for j in self._jobs.values() + if j.session_id == session_id and j.status != "running"), + key=lambda j: j.finished_at or 0, + ) + for job in finished[: count - self.FINISHED_CAP_PER_SESSION]: + del self._jobs[job.task_id] + reaped += 1 + return reaped + + +_manager: TaskManager | None = None + + +def get_task_manager() -> TaskManager: + global _manager + if _manager is None: + _manager = TaskManager() + return _manager + + +def args_summary(args: dict, limit: int = 200) -> str: + """Compact human-readable argument summary for task ids / notes.""" + try: + return json.dumps(args, ensure_ascii=False)[:limit] + except (TypeError, ValueError): + return str(args)[:limit] \ No newline at end of file diff --git a/navi/core/tool_executor.py b/navi/core/tool_executor.py index a71be58..ad47aee 100644 --- a/navi/core/tool_executor.py +++ b/navi/core/tool_executor.py @@ -1,12 +1,13 @@ """Tool execution helpers — extracted from agent.py.""" import asyncio +import json from typing import TYPE_CHECKING import structlog from navi.llm.base import Message, ToolCallRequest -from navi.tools._internal.base import Tool +from navi.tools._internal.base import Tool, ToolResult, current_session_id if TYPE_CHECKING: from navi.core.events import ToolEvent @@ -14,6 +15,12 @@ log = structlog.get_logger() +def _backgroundable_tools() -> set[str]: + from navi.config import settings + + return {t.strip() for t in settings.backgroundable_tools.split(",") if t.strip()} + + def _resolve_tool(tool_map: dict[str, Tool], name: str) -> tuple[str, Tool | None]: """Resolve exact tool names plus common MCP alias mistakes.""" from navi.mcp.tools import is_mcp_tool, parse_mcp_name @@ -77,6 +84,72 @@ def __init__(self, tool_registry) -> None: self._tools = tool_registry + def _maybe_background( + self, + tc: ToolCallRequest, + resolved_name: str, + tool: Tool | None, + ctx=None, + ) -> ToolResult | None: + """Detach the call into the TaskManager when args carry background=true. + + Returns None when the call should run inline (no flag, unknown tool, + tool not backgroundable). Otherwise pops the flag from a copy of the + arguments, submits the tool execution to the TaskManager and returns + the immediate ToolResult the agent sees. + """ + args = dict(tc.arguments or {}) + want_background = args.pop("background", None) + if not want_background or tool is None: + return None + if resolved_name not in _backgroundable_tools(): + return None + # terminal open(background=true) has its own native meaning — start a + # persistent detached process. Never hijack it. + if resolved_name == "terminal" and args.get("action") == "open": + return None + + from navi.core.tasks import args_summary, get_task_manager + + session_id = getattr(ctx, "session_id", None) or current_session_id.get() + coro_factory = lambda bg_ctx: tool.execute(args, ctx=bg_ctx) # noqa: E731 + # Subagents emit a dense event stream — give them a deeper ring. + ring_size = 200 if resolved_name == "spawn_agent" else None + job = get_task_manager().submit( + session_id=session_id, + tool=resolved_name, + args=args, + coro_factory=coro_factory, + ctx=ctx, + parent_tool_call_id=tc.id, + ring_size=ring_size, + ) + if isinstance(job, str): + # Cap/rate-limit hit — surface as a normal failed result so the + # agent can fall back to running the tool inline. + return ToolResult( + success=False, + output=f"Cannot run in background: {job}. Run it in the foreground instead.", + error="task_cap", + ) + output = json.dumps( + { + "task_id": job.task_id, + "tool": resolved_name, + "status": "running", + "args_summary": args_summary(args), + "hint": ( + "The call is running in the background — continue with other work. " + "Use tasks check/wait with this task_id to collect the result; " + "it will also arrive as an automatic [Background task results] " + "note next turn." + ), + }, + ensure_ascii=False, + ) + return ToolResult(success=True, output=output, + metadata={"background": True, "task_id": job.task_id}) + async def _execute_one( self, tc: ToolCallRequest, @@ -103,13 +176,23 @@ tool_call_id=tc.id, ) else: - log.info("tool.execute", tool=resolved_name, requested_tool=tc.name, args=tc.arguments) + background_result = self._maybe_background(tc, resolved_name, tool, ctx) middlewares = getattr(self._tools, "_middlewares", []) - for mw in middlewares: - await mw.before_execute(resolved_name, tc.arguments) - result = await tool.execute(tc.arguments, ctx=ctx) - for mw in middlewares: - await mw.after_execute(resolved_name, tc.arguments, result) + if background_result is not None: + log.info( + "tool.backgrounded", tool=resolved_name, + args=tc.arguments, task_id=background_result.metadata.get("task_id"), + ) + result = background_result + for mw in middlewares: + await mw.after_execute(resolved_name, tc.arguments, result) + else: + log.info("tool.execute", tool=resolved_name, requested_tool=tc.name, args=tc.arguments) + for mw in middlewares: + await mw.before_execute(resolved_name, tc.arguments) + result = await tool.execute(tc.arguments, ctx=ctx) + for mw in middlewares: + await mw.after_execute(resolved_name, tc.arguments, result) content = result.to_message_content() metadata = result.metadata or {} event = ToolEvent( diff --git a/navi/profiles/base.py b/navi/profiles/base.py index 7f7340a..09c68af 100644 --- a/navi/profiles/base.py +++ b/navi/profiles/base.py @@ -99,6 +99,10 @@ # OFF = legacy "free flight" behavior (explore broadly, finish discovered work). scope_boundary_enabled: bool = False + # Execute multiple tool calls from one LLM response concurrently instead of + # one-by-one. None = inherit the global settings.parallel_tool_calls. + parallel_tool_calls: bool | None = None + # Detect when the model is looping without todo progress for # anti_stall_threshold iterations and inject a hard warning. anti_stall_enabled: bool = True diff --git a/navi/tools/__init__.py b/navi/tools/__init__.py index fe8ea1c..4136696 100644 --- a/navi/tools/__init__.py +++ b/navi/tools/__init__.py @@ -17,6 +17,7 @@ from .reflect import ReflectTool from .plan import PlanRunner, PlanTool from .peer import PeerTool +from .tasks import TasksTool __all__ = [ "Tool", @@ -32,6 +33,7 @@ "ScheduleRecallTool", "TestMcpToolTool", "SpawnAgentTool", + "TasksTool", "TodoTool", "ScratchpadTool", "SwitchProfileTool", diff --git a/navi/tools/code_exec.py b/navi/tools/code_exec.py index 3b9c5ce..013e149 100644 --- a/navi/tools/code_exec.py +++ b/navi/tools/code_exec.py @@ -100,6 +100,14 @@ "Raise it for long computations or test suites." ), }, + "background": { + "type": "boolean", + "description": ( + "Detach the call into a background task — returns a task_id immediately, " + "result arrives via the tasks tool / completion note. Use for long " + "computations/test suites. Default false." + ), + }, }, "required": ["code"], } diff --git a/navi/tools/peer.py b/navi/tools/peer.py index 0d838d0..96f2791 100644 --- a/navi/tools/peer.py +++ b/navi/tools/peer.py @@ -80,6 +80,14 @@ "type": "string", "description": "The question for 'ask' - concrete, self-contained, answerable by the peer on its machine", }, + "background": { + "type": "boolean", + "description": ( + "For 'ask': detach into a background task — returns a task_id immediately " + "(peers can take up to a couple of minutes); collect the answer via the " + "tasks tool / completion note. Default false." + ), + }, }, "required": ["action"], } diff --git a/navi/tools/spawn_agent.py b/navi/tools/spawn_agent.py index f7c9b30..b6b699a 100644 --- a/navi/tools/spawn_agent.py +++ b/navi/tools/spawn_agent.py @@ -22,8 +22,10 @@ "CRITICAL: one spawn_agent call = one plan step. " "If your plan has three AGENT steps, you make three separate spawn_agent calls — " "one per step. Never bundle multiple plan steps into a single sub-agent.\n\n" - "SYNCHRONOUS — blocks until the sub-agent fully completes. " - "There is no background process and no continuation.\n\n" + "SYNCHRONOUS by default — blocks until the sub-agent fully completes. " + "For long research/ops sub-agents set background=true: the call returns " + "a task_id immediately, the sub-agent runs detached, and the result " + "arrives via the tasks tool and an automatic completion note.\n\n" "USER CANNOT SEE sub-agent output — synthesise findings into your own response.\n\n" "USE when a step requires 3+ tool calls to complete as a single logical unit. " "DO NOT USE for a single tool call — call the tool directly.\n\n" @@ -78,6 +80,15 @@ "type": "integer", "description": "Maximum tool-call iterations for the sub-agent (default: 40).", }, + "background": { + "type": "boolean", + "description": ( + "Run detached: returns a task_id immediately and the sub-agent keeps " + "working while you continue. Collect via tasks check/wait; the result " + "also arrives as an automatic note next turn. Use for sub-agents expected " + "to run longer than ~45-60s. Default false (synchronous)." + ), + }, "inherit_system_prompt": { "type": "boolean", "description": ( diff --git a/navi/tools/ssh_exec.py b/navi/tools/ssh_exec.py index 3294ad6..8014188 100644 --- a/navi/tools/ssh_exec.py +++ b/navi/tools/ssh_exec.py @@ -187,6 +187,14 @@ "type": "integer", "description": f"Timeout in seconds (default {_TIMEOUT})", }, + "background": { + "type": "boolean", + "description": ( + "Detach the call into a background task — returns a task_id immediately, " + "result arrives via the tasks tool / completion note. Use for long remote " + "commands (builds, package installs, backups). Default false." + ), + }, }, "required": [], } diff --git a/navi/tools/tasks.py b/navi/tools/tasks.py new file mode 100644 index 0000000..12798a7 --- /dev/null +++ b/navi/tools/tasks.py @@ -0,0 +1,161 @@ +"""Background-task management tool — list/check/wait/cancel detached jobs.""" +from __future__ import annotations + +import asyncio + +from navi.tools._internal.base import ( + Tool, + ToolContext, + ToolResult, + current_session_id, +) + +_MAX_CHECK_OUTPUT = 2000 +_MAX_WAIT_TIMEOUT = 120.0 + + +def _sid(ctx: ToolContext | None) -> str | None: + return (ctx.session_id if ctx else None) or current_session_id.get() + + +class TasksTool(Tool): + name = "tasks" + description = ( + "Manage background tasks started with background=true on long-running " + "tool calls (terminal, ssh_exec, peer ask, spawn_agent, code_exec). " + "list — show this session's tasks; " + "check — status + result + recent progress of one task; " + "wait — block until a task finishes (timeout cap 120s) and return its result; " + "cancel — stop a running task. " + "Prefer check + continue working over wait: wait blocks your turn." + ) + parameters = { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["list", "check", "wait", "cancel"], + "description": "What to do with background tasks.", + }, + "task_id": { + "type": "string", + "description": "Task id (bt-...) — required for check/wait/cancel.", + }, + "timeout": { + "type": "number", + "description": "Max seconds to wait in the 'wait' action (default 120).", + }, + }, + "required": ["action"], + } + + async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: + from navi.core.tasks import get_task_manager + + action = params.get("action") + manager = get_task_manager() + session_id = _sid(ctx) + + if action == "list": + jobs = manager.list(session_id or "__default__") + if not jobs: + return ToolResult(success=True, output="No background tasks in this session.") + lines = [ + f"{j.task_id} {j.tool:<12} {j.status:<10} {self._age(j)}s" + + (f" {j.preview(limit=120)}" if j.status != "running" else "") + for j in jobs + ] + return ToolResult(success=True, output="\n".join(lines)) + + task_id = params.get("task_id") + if not task_id: + return ToolResult(success=False, output="", error="'task_id' is required") + job = manager.get(task_id, session_id or "__default__") + if job is None: + return ToolResult( + success=False, + output=f"Task '{task_id}' not found in this session (it may have been reaped).", + error="task_not_found", + ) + + if action == "check": + return ToolResult(success=True, output=self._render_check(job)) + + if action == "wait": + timeout = min(float(params.get("timeout") or _MAX_WAIT_TIMEOUT), _MAX_WAIT_TIMEOUT) + try: + await asyncio.wait_for(job.done.wait(), timeout=timeout) + except asyncio.TimeoutError: + return ToolResult( + success=False, + output=( + f"Task {job.task_id} is still running after {timeout:g}s " + f"(status: {job.status}). Continue other work and check again later." + ), + error="wait_timeout", + ) + return ToolResult(success=True, output=self._render_check(job)) + + if action == "cancel": + if manager.cancel(job): + return ToolResult(success=True, output=f"Task {job.task_id} cancelled.") + return ToolResult( + success=False, + output=f"Task {job.task_id} is not running (status: {job.status}).", + error="not_running", + ) + + return ToolResult(success=False, output="", error=f"Unknown action '{action}'") + + # -- rendering ---------------------------------------------------------- + + @staticmethod + def _age(job) -> int: + import time + + end = job.finished_at if job.finished_at is not None else time.time() + return int(end - job.created_at) + + def _render_check(self, job) -> str: + parts = [ + f"Task {job.task_id} ({job.tool}) — status: {job.status}, age: {self._age(job)}s", + ] + if job.subagent_tokens is not None: + parts.append(f"Sub-agent tokens: {job.subagent_tokens}") + if job.status == "running": + progress = self._recent_events(job) + if progress: + parts.append("Recent progress:\n" + progress) + else: + parts.append("No progress events yet.") + else: + if job.result is not None: + out = job.result.to_message_content() + if len(out) > _MAX_CHECK_OUTPUT: + out = out[:_MAX_CHECK_OUTPUT] + "\n... (truncated — full result is in the completion note)" + parts.append(("Result:\n" if job.result.success else "Result (failed):\n") + out) + if job.error and job.result is None: + parts.append(f"Error: {job.error}") + return "\n".join(parts) + + @staticmethod + def _recent_events(job, limit: int = 8) -> str: + """Compact summary of the last events in the task's ring buffer.""" + if job.ring is None: + return "" + items = list(job.ring._queue)[-limit:] + lines = [] + for item in items: + kind = type(item).__name__ + detail = "" + if kind == "ToolStarted" or kind == "ToolEvent": + detail = getattr(item, "tool_name", "") + if kind == "ToolEvent": + detail += " ✓" if getattr(item, "success", False) else " ✗" + elif kind == "TurnThinking": + text = (getattr(item, "thinking", "") or "").strip().replace("\n", " ") + detail = text[:100] + elif kind == "PlanReady": + detail = "plan ready" + lines.append(f" {kind}{(': ' + detail) if detail else ''}") + return "\n".join(lines) if lines else "" \ No newline at end of file diff --git a/navi/tools/terminal.py b/navi/tools/terminal.py index 1582e19..cec1953 100644 --- a/navi/tools/terminal.py +++ b/navi/tools/terminal.py @@ -143,7 +143,12 @@ }, "background": { "type": "boolean", - "description": "Run in background without waiting for completion (for open). Default false.", + "description": ( + "action=open: start the persistent terminal detached (native mode). " + "action=run: detach the whole call into a background task — returns a " + "task_id immediately, result arrives via the tasks tool / completion note. " + "Use for builds, tests, installs — anything slower than ~45-60s. Default false." + ), }, "working_dir": { "type": "string", diff --git a/persona.txt b/persona.txt index 3ba5c54..ca1ae43 100644 --- a/persona.txt +++ b/persona.txt @@ -57,7 +57,7 @@ - When complete or fundamentally blocked, report the outcome once, concisely. SUB-AGENTS: -spawn_agent is synchronous and blocking — when it returns, the sub-agent has fully completed. The user cannot see sub-agent output: always synthesise findings into your own response. +spawn_agent without `background: true` is synchronous and blocking — when it returns, the sub-agent has fully completed. With `background: true` the call detaches immediately and returns a `task_id` (see PARALLELISM). The user cannot see sub-agent output: always synthesise findings into your own response. ONE PLAN STEP = ONE spawn_agent CALL. If your plan has four AGENT steps, you make four separate calls — one per step. Never pass your full plan to a single subagent. @@ -74,6 +74,15 @@ If the result says "hit iteration limit" — it may be incomplete. Note what is missing in your response. +PARALLELISM AND BACKGROUND TASKS: +You can keep working while long operations run. Tools `terminal` (run), `ssh_exec`, `peer`, `spawn_agent` and `code_exec` accept `"background": true` — the call detaches immediately and returns a `task_id` (plus a `tasks` tool to manage them). Rules: +- Detach anything expected to take longer than ~45-60 seconds: a long terminal command, an ssh job, a peer ask, a background sub-agent. While it runs, do useful work in the foreground. +- Discipline: start task → keep working → `tasks` action `check` occasionally (do not spam it) → `tasks` action `wait` ONLY when the next step needs the result. Results of finished tasks are also injected as a note at the start of your next turn. +- Keep at most 2-3 background tasks at once. Background sub-agents are capped even lower. +- Never start new background tasks merely because a task-results note arrived — that note is a report, not a request. +- Use `tasks` action `cancel` to kill a task that is no longer needed. Stopping a run does NOT stop background tasks. +- When the profile allows parallel tool calls, you may put several independent tool calls in one batch: they start together and results are reported in call order. Batch only independent calls; NEVER put two actions on the same terminal (or the same ssh host state) into one batch. + REFLECTION: You have a reflect tool: Critic (challenges assumptions, surfaces risks), Pragmatist (finds simplest path), Detailer (spots missing requirements). All three run in parallel — it is fast. diff --git a/persona_navi_code.txt b/persona_navi_code.txt index 11b6eb8..3b342a9 100644 --- a/persona_navi_code.txt +++ b/persona_navi_code.txt @@ -13,6 +13,13 @@ - Для быстрых проверок кода — `code_exec`. - Перед созданием нового инструмента всегда читай `tool_manual("write_tool")`. +Параллельность и фоновые задачи: +- Всё, что дольше ~45-60 секунд (длинные команды в `terminal`, `ssh_exec`, `peer`, `spawn_agent`, `code_exec`), запускай с `"background": true` — получишь `task_id` и сможешь продолжать работу. +- Дисциплина: запустил → полезная работа → `tasks check` (не спамить) → `tasks wait` только когда результат нужен для следующего шага. Результаты завершённых задач придут нотой в начале следующего тёрна. +- Держи не более 2-3 фоновых задач одновременно. Не спавни новые задачи в ответ на ноту результатов — это отчёт, а не запрос. +- При параллельных вызовах тулов батчи только независимые вызовы; никогда не ставь два действия над одним терминалом в один батч. +- `tasks cancel` убивает фоновую задачу; остановка рана её не останавливает. + Язык общения: Используй тот язык, на котором к тебе обратился пользователь (по умолчанию русский). ПОЛ И РОД (строго): Ты — женщина. На русском языке ты ВСЕГДА говоришь о себе в женском роде: «я сделала», «я нашла», «я готова», «я решила», «я проверила», «должна». Никогда «я сделал», «я нашёл», «я готов», «должен» — ни в ответах, ни в промежуточных рассуждениях. Правило без исключений. diff --git a/pyproject.toml b/pyproject.toml index 3f32775..45f29dd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,8 +41,9 @@ "gnexus-gauth @ git+https://git.gnexus.space/root/gnexus-auth-client-py.git", "cryptography>=42", - # MCP (Model Context Protocol) - "mcp>=1.27", + # MCP (Model Context Protocol). v2 renamed FastMCP → MCPServer with + # breaking API changes — pin to v1 until navi/mcp is migrated. + "mcp>=1.27,<2", # Config "pydantic>=2.7", diff --git a/tests/unit/api/test_websocket.py b/tests/unit/api/test_websocket.py index 7c6e137..4851791 100644 --- a/tests/unit/api/test_websocket.py +++ b/tests/unit/api/test_websocket.py @@ -191,8 +191,8 @@ # ── Concurrent run guard ───────────────────────────────────────────────────── @pytest.mark.anyio -async def test_concurrent_run_guard_rejects_second_message(mock_websocket, mock_session, mock_user, monkeypatch): - """Sending a second message while a run is active yields a WebSocket error.""" +async def test_concurrent_run_guard_queues_second_message(mock_websocket, mock_session, mock_user, monkeypatch): + """Sending a second message while a run is active queues it (message_queued).""" monkeypatch.setattr(ws_mod, "get_current_user_ws", AsyncMock(return_value=mock_user)) mock_store = MagicMock() mock_store.get = AsyncMock(return_value=mock_session) @@ -231,9 +231,14 @@ await ws_mod.websocket_session("s1", mock_websocket) calls = [c.args[0] for c in mock_websocket.send_json.call_args_list] - error_calls = [c for c in calls if c["type"] == "error"] - assert len(error_calls) == 1 - assert "already running" in error_calls[0]["message"] + queued_calls = [c for c in calls if c["type"] == "message_queued"] + assert len(queued_calls) == 1 + assert queued_calls[0]["position"] == 1 + assert queued_calls[0]["queue_len"] == 1 + # No error was sent for the queued message + assert not [c for c in calls if c["type"] == "error"] + # The message is retained in the session queue + assert orchestrator.has_pending("s1") # Cleanup background task state = orchestrator._sessions.get("s1") @@ -527,3 +532,102 @@ mock_websocket.close.assert_awaited_with(code=4003, reason="Access denied") assert "s1" not in orchestrator._sessions + + +# ── Message queue (busy → message_queued → drain) ─────────────────────────── + + +@pytest.mark.anyio +async def test_drain_runs_queued_messages_after_run(mock_websocket, mock_user, monkeypatch): + """After a run finishes on the socket, queued messages execute back-to-back.""" + orchestrator = AgentSessionOrchestrator(MagicMock()) + ran = [] + + async def fake_single(**kwargs): + ran.append(kwargs["user_content"]) + return True, True + + monkeypatch.setattr(ws_mod, "_run_single_message", AsyncMock(side_effect=fake_single)) + orchestrator.queue_message("s1", { + "user_content": "queued one", + "raw_images": ["img"], + "display_content": "queued one", + "uploaded_files": [], + "hidden": False, + "user": mock_user, + "cwd": "/tmp", + }) + + connected = await ws_mod._start_agent_run( + session_id="s1", user_content="original", raw_images=None, + display_content="original", uploaded_files=[], hidden=False, + websocket=mock_websocket, orchestrator=orchestrator, + session_store=MagicMock(), user=mock_user, + ) + + assert connected is True + assert ran == ["original", "queued one"] + # drained entry preserved identity and fields + second = ws_mod._run_single_message.call_args_list[1].kwargs + assert second["raw_images"] == ["img"] + assert second["user"] is mock_user + assert second["cwd"] == "/tmp" + assert not orchestrator.has_pending("s1") + + +@pytest.mark.anyio +async def test_drain_stops_when_socket_dies(mock_websocket, mock_user, monkeypatch): + """A disconnect mid-drain leaves the remaining queue intact.""" + orchestrator = AgentSessionOrchestrator(MagicMock()) + ran = [] + + async def fake_single(**kwargs): + ran.append(kwargs["user_content"]) + # the second drained message "disconnects" the socket + return (False, True) if kwargs["user_content"] == "second" else (True, True) + + monkeypatch.setattr(ws_mod, "_run_single_message", AsyncMock(side_effect=fake_single)) + orchestrator.queue_message("s1", {"user_content": "second", "uploaded_files": []}) + orchestrator.queue_message("s1", {"user_content": "third", "uploaded_files": []}) + + connected = await ws_mod._start_agent_run( + session_id="s1", user_content="first", raw_images=None, + display_content="first", uploaded_files=[], hidden=False, + websocket=mock_websocket, orchestrator=orchestrator, + session_store=MagicMock(), user=mock_user, + ) + + assert connected is False + assert ran == ["first", "second"] + assert orchestrator.pop_pending("s1")["user_content"] == "third" + + +@pytest.mark.anyio +async def test_queued_entry_keeps_user_identity(mock_websocket, monkeypatch): + """Messages queued from another socket run with the queueing user.""" + from navi.auth import User + + orchestrator = AgentSessionOrchestrator(MagicMock()) + other = User(id="user-b", email="b@test.com", role="user") + ws_a = AsyncMock() + ws_a.send_json = AsyncMock() + + # Simulate socket B queueing while a run is "active" + orchestrator.create_run("s1") + connected, ran = await ws_mod._run_single_message( + session_id="s1", user_content="from b", raw_images=None, + display_content="from b", uploaded_files=[], hidden=False, + websocket=ws_a, orchestrator=orchestrator, + session_store=MagicMock(), user=other, + ) + assert connected is True + sent = ws_a.send_json.call_args_list[0].args[0] + assert sent["type"] == "message_queued" + + entry = orchestrator.pop_pending("s1") + assert entry["user"] is other + assert entry["user_content"] == "from b" + state = orchestrator._sessions.get("s1") + if state and state.run: + state.run = None + orchestrator._sessions.pop("s1", None) diff --git a/tests/unit/core/test_orchestrator_queue.py b/tests/unit/core/test_orchestrator_queue.py new file mode 100644 index 0000000..3e791c0 --- /dev/null +++ b/tests/unit/core/test_orchestrator_queue.py @@ -0,0 +1,120 @@ +"""Unit tests for the user message queue in AgentSessionOrchestrator (Q.1).""" + +import asyncio +from unittest.mock import MagicMock + +import pytest + +from navi.core.orchestrator import AgentSessionOrchestrator + + +@pytest.fixture +def orchestrator(): + container = MagicMock() + return AgentSessionOrchestrator(container) + + +class TestQueueMessage: + def test_fifo_order_and_positions(self, orchestrator): + r1 = orchestrator.queue_message("s1", {"user_content": "first"}) + r2 = orchestrator.queue_message("s1", {"user_content": "second"}) + assert r1 == {"position": 1, "queue_len": 1, "dropped_total": 0} + assert r2 == {"position": 2, "queue_len": 2, "dropped_total": 0} + assert orchestrator.pop_pending("s1")["user_content"] == "first" + assert orchestrator.pop_pending("s1")["user_content"] == "second" + + def test_overflow_drops_oldest_and_counts(self, orchestrator): + for i in range(7): # maxlen=5 by default + orchestrator.queue_message("s1", {"user_content": f"m{i}"}) + entry = orchestrator.pop_pending("s1") + assert entry["user_content"] == "m2" # m0, m1 dropped + info = orchestrator.queue_message("s1", {"user_content": "more"}) + assert info["dropped_total"] == 2 + + def test_sessions_are_isolated(self, orchestrator): + orchestrator.queue_message("s1", {"user_content": "a"}) + assert not orchestrator.has_pending("s2") + assert orchestrator.pop_pending("s2") is None + + def test_has_pending(self, orchestrator): + assert not orchestrator.has_pending("s1") + orchestrator.queue_message("s1", {"user_content": "a"}) + assert orchestrator.has_pending("s1") + orchestrator.pop_pending("s1") + assert not orchestrator.has_pending("s1") + + +class TestHeadlessDrain: + async def test_schedule_skips_when_no_pending(self, orchestrator, monkeypatch): + started = [] + + async def fake_drain(session_id, store): + started.append(session_id) + + monkeypatch.setattr(orchestrator, "_drain_queued", fake_drain) + orchestrator.schedule_queued_drain("s1", None) + assert started == [] + + async def test_schedule_skips_when_websocket_watching(self, orchestrator, monkeypatch): + started = [] + + async def fake_drain(session_id, store): + started.append(session_id) + + monkeypatch.setattr(orchestrator, "_drain_queued", fake_drain) + orchestrator.queue_message("s1", {"user_content": "a"}) + orchestrator._get_or_create_state("s1").websockets.append(MagicMock()) + orchestrator.schedule_queued_drain("s1", None) + assert started == [] + + async def test_schedule_runs_headless_drain(self, orchestrator, monkeypatch): + started = [] + + async def fake_drain(session_id, store): + started.append(session_id) + + monkeypatch.setattr(orchestrator, "_drain_queued", fake_drain) + orchestrator.queue_message("s1", {"user_content": "a"}) + orchestrator.schedule_queued_drain("s1", None) + await asyncio.sleep(0) + assert started == ["s1"] + + async def test_drain_runs_queued_messages_in_order(self, orchestrator, monkeypatch): + calls = [] + + async def fake_run_agent(session_id, content, *args, **kwargs): + calls.append(content) + # mirror the real run_agent finally: clear the run so the next + # queued message is not re-queued by the busy guard + state = orchestrator._sessions.get(session_id) + if state is not None: + state.run = None + + monkeypatch.setattr(orchestrator, "run_agent", fake_run_agent) + orchestrator.queue_message("s1", {"user_content": "first"}) + orchestrator.queue_message("s1", {"user_content": "second"}) + await orchestrator._drain_queued("s1", None) + assert calls == ["first", "second"] + assert not orchestrator.has_pending("s1") + + async def test_drain_requeues_when_run_takes_over(self, orchestrator, monkeypatch): + calls = [] + + async def fake_run_agent(session_id, content, *args, **kwargs): + calls.append(content) + + monkeypatch.setattr(orchestrator, "run_agent", fake_run_agent) + monkeypatch.setattr(orchestrator, "is_running", lambda sid: True) + orchestrator.queue_message("s1", {"user_content": "held"}) + await orchestrator._drain_queued("s1", None) + assert calls == [] + assert orchestrator.has_pending("s1") # put back for the active run + + async def test_drain_swallows_run_agent_failure(self, orchestrator, monkeypatch): + async def failing_run_agent(*args, **kwargs): + raise RuntimeError("llm down") + + monkeypatch.setattr(orchestrator, "run_agent", failing_run_agent) + orchestrator.queue_message("s1", {"user_content": "a"}) + await orchestrator._drain_queued("s1", None) # must not raise + # the failed message is not re-queued (no infinite retry loop) \ No newline at end of file diff --git a/tests/unit/core/test_parallel_tools.py b/tests/unit/core/test_parallel_tools.py new file mode 100644 index 0000000..c6bee9d --- /dev/null +++ b/tests/unit/core/test_parallel_tools.py @@ -0,0 +1,228 @@ +"""Unit tests for the parallel tool-call batch (Ф3).""" + +import asyncio +import time +from types import SimpleNamespace + +import pytest + +from navi.core.agent import Agent, AgentTurnContext +from navi.core.events import TextDelta, ToolEvent, ToolStarted +from navi.core.tool_executor import ToolExecutor +from navi.llm.base import ToolCallRequest +from navi.tools._internal.base import ToolResult, current_event_sink + + +class FakeStore: + def __init__(self): + self.saves = 0 + + async def save(self, session): + self.saves += 1 + + +class SlowTool: + """Sleeps, optionally emits a live event, returns output.""" + + def __init__(self, name, delay, emit=None, fail=False): + self.name = name + self._delay = delay + self._emit = emit + self._fail = fail + + async def execute(self, arguments, ctx=None): + await asyncio.sleep(self._delay) + if self._emit is not None: + sink = current_event_sink.get() + await sink.put(self._emit) + if self._fail: + raise ValueError(f"{self.name} blew up") + return ToolResult(success=True, output=f"done {self.name}") + + +def make_agent(): + agent = object.__new__(Agent) + agent._tool_executor = ToolExecutor(SimpleNamespace(_middlewares=[])) + agent._sessions = FakeStore() + return agent, agent._sessions + + +def make_tcs(*names): + return [ToolCallRequest(id=f"tc-{i}", name=n, arguments={}) + for i, n in enumerate(names)] + + +def make_turn_ctx(parallel=True): + return AgentTurnContext(turn_start=time.monotonic(), parallel_tool_calls=parallel) + + +class TestParallelBatch: + async def test_wall_clock_shorter_than_sequential(self): + agent, _ = make_agent() + tools = [SlowTool("a", 0.25), SlowTool("b", 0.25)] + tool_map = {t.name: t for t in tools} + + async def run(): + return [ev async for ev in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, make_turn_ctx(), SimpleNamespace(messages=[], context=[]), + None, None)] + + start = time.monotonic() + events = await run() + elapsed = time.monotonic() - start + assert elapsed < 0.45 # sequential would take ≥0.5s + tool_events = [e for e in events if isinstance(e, ToolEvent)] + assert len(tool_events) == 2 + + async def test_tool_started_all_before_any_result(self): + agent, _ = make_agent() + tools = [SlowTool("a", 0.25), SlowTool("b", 0.25)] + events = [] + async for ev in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, make_turn_ctx(), + SimpleNamespace(messages=[], context=[]), None, None, + ): + events.append(ev) + started = [i for i, e in enumerate(events) if isinstance(e, ToolStarted)] + done = [i for i, e in enumerate(events) if isinstance(e, ToolEvent)] + assert started == [0, 1] + assert done == [2, 3] + + async def test_results_in_call_order_despite_finish_order(self): + agent, _ = make_agent() + # second tool finishes first + tools = [SlowTool("slow", 0.2), SlowTool("fast", 0.01)] + events = [ev async for ev in agent._execute_tools_parallel( + make_tcs("slow", "fast"), tools, make_turn_ctx(), + SimpleNamespace(messages=[], context=[]), None, None)] + tool_events = [e for e in events if isinstance(e, ToolEvent)] + assert [e.tool_name for e in tool_events] == ["slow", "fast"] + + async def test_live_events_merged_through_shared_queue(self): + agent, _ = make_agent() + tools = [SlowTool("a", 0.05, emit=TextDelta(delta="A")), + SlowTool("b", 0.05, emit=TextDelta(delta="B"))] + events = [ev async for ev in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, make_turn_ctx(), + SimpleNamespace(messages=[], context=[]), None, None)] + deltas = [e.delta for e in events if isinstance(e, TextDelta)] + assert sorted(deltas) == ["A", "B"] + + async def test_one_failure_does_not_kill_neighbours(self): + agent, store = make_agent() + tools = [SlowTool("bad", 0.01, fail=True), SlowTool("good", 0.01)] + session = SimpleNamespace(messages=[], context=[]) + events = [ev async for ev in agent._execute_tools_parallel( + make_tcs("bad", "good"), tools, make_turn_ctx(), session, None, None)] + tool_events = [e for e in events if isinstance(e, ToolEvent)] + by_name = {e.tool_name: e for e in tool_events} + assert by_name["bad"].success is False + assert by_name["good"].success is True + # both tool messages recorded, in call order + assert [m.name for m in session.messages] == ["bad", "good"] + + async def test_stop_mid_batch_synthesises_results_in_order(self): + agent, store = make_agent() + tools = [SlowTool("a", 5.0), SlowTool("b", 5.0)] + session = SimpleNamespace(messages=[], context=[]) + stop_event = asyncio.Event() + + async def set_stop(): + await asyncio.sleep(0.1) + stop_event.set() + + asyncio.create_task(set_stop()) + events = [ev async for ev in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, make_turn_ctx(), session, stop_event, None)] + tool_events = [e for e in events if isinstance(e, ToolEvent)] + assert len(tool_events) == 2 + assert all(not e.success for e in tool_events) + assert all("stopped by the user" in e.result for e in tool_events) + assert all(m.is_context is False for m in session.messages) + + async def test_single_save_per_batch(self): + agent, store = make_agent() + tools = [SlowTool("a", 0.01), SlowTool("b", 0.01)] + session = SimpleNamespace(messages=[], context=[]) + async for _ in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, make_turn_ctx(), session, None, None): + pass + assert store.saves == 1 + + async def test_tool_call_count_incremented(self): + agent, _ = make_agent() + tools = [SlowTool("a", 0.01), SlowTool("b", 0.01)] + turn_ctx = make_turn_ctx() + async for _ in agent._execute_tools_parallel( + make_tcs("a", "b"), tools, turn_ctx, + SimpleNamespace(messages=[], context=[]), None, None): + pass + assert turn_ctx.tool_call_count == 2 + + +class TestDispatch: + async def test_gate_off_uses_sequential_path(self): + """parallel_tool_calls=False keeps the strict sequential ordering.""" + agent, _ = make_agent() + tools = [SlowTool("a", 0.05), SlowTool("b", 0.05)] + events = [ev async for ev in agent._execute_tools_with_sink( + make_tcs("a", "b"), tools, make_turn_ctx(parallel=False), + SimpleNamespace(messages=[], context=[]), None, None)] + kinds = [type(e).__name__ for e in events] + # sequential: Started(a), Event(a), Started(b), Event(b) + assert kinds == ["ToolStarted", "ToolEvent", "ToolStarted", "ToolEvent"] + + async def test_gate_on_routes_to_parallel(self): + agent, _ = make_agent() + tools = [SlowTool("a", 0.25), SlowTool("b", 0.25)] + events = [] + async for ev in agent._execute_tools_with_sink( + make_tcs("a", "b"), tools, make_turn_ctx(parallel=True), + SimpleNamespace(messages=[], context=[]), None, None): + events.append(ev) + kinds = [type(e).__name__ for e in events] + assert kinds[:2] == ["ToolStarted", "ToolStarted"] + + async def test_single_call_batch_stays_sequential(self): + """One tool call never needs the parallel machinery.""" + agent, _ = make_agent() + tools = [SlowTool("a", 0.01)] + events = [ev async for ev in agent._execute_tools_with_sink( + make_tcs("a"), tools, make_turn_ctx(parallel=True), + SimpleNamespace(messages=[], context=[]), None, None)] + kinds = [type(e).__name__ for e in events] + assert kinds == ["ToolStarted", "ToolEvent"] + + +class TestRepair: + def test_repair_dangling_tool_calls(self): + from navi.core.pg_session_store import repair_dangling_tool_calls + from navi.llm.base import Message + + assistant = Message(role="assistant", content=None, + tool_calls=[ToolCallRequest(id="t1", name="a", arguments={}), + ToolCallRequest(id="t2", name="b", arguments={})]) + answered = Message(role="tool", content="ok", tool_call_id="t1", name="a") + later = Message(role="user", content="next") + messages = [assistant, answered, later] + + added = repair_dangling_tool_calls(messages) + assert added == 1 + # placeholder sits after the answered tool message, before the user msg + assert messages[2].role == "tool" + assert messages[2].tool_call_id == "t2" + assert messages[2].content == "[Interrupted before result]" + assert messages[2].is_context is False + assert messages[3] is later + + def test_repair_noop_when_complete(self): + from navi.core.pg_session_store import repair_dangling_tool_calls + from navi.llm.base import Message + + messages = [ + Message(role="assistant", content=None, + tool_calls=[ToolCallRequest(id="t1", name="a", arguments={})]), + Message(role="tool", content="ok", tool_call_id="t1", name="a"), + ] + assert repair_dangling_tool_calls(messages) == 0 + assert len(messages) == 2 \ No newline at end of file diff --git a/tests/unit/core/test_task_manager.py b/tests/unit/core/test_task_manager.py new file mode 100644 index 0000000..3308dec --- /dev/null +++ b/tests/unit/core/test_task_manager.py @@ -0,0 +1,258 @@ +"""Unit tests for navi.core.tasks — TaskManager / TaskJob / BoundedEventQueue.""" + +import asyncio +import time + +import pytest + +from navi.core import tasks as tasks_mod +from navi.core.events import TaskUpdate +from navi.core.tasks import ( + BoundedEventQueue, + TaskJob, + TaskManager, + args_summary, + get_task_manager, +) +from navi.tools._internal.base import ToolContext, ToolResult, current_stop_event + + +def patch_settings(monkeypatch, **overrides): + """Replace the frozen settings singleton with an overridden copy.""" + import navi.config as config_mod + from navi.config import Settings + + new_settings = Settings(**overrides) + monkeypatch.setattr(config_mod, "settings", new_settings) + return new_settings + + +@pytest.fixture(autouse=True) +def fresh_manager(monkeypatch): + """Isolate the module singleton; clean up stray tasks afterwards.""" + manager = TaskManager() + monkeypatch.setattr(tasks_mod, "_manager", manager) + yield manager + for job in list(manager._jobs.values()): + if job.task is not None and not job.task.done(): + job.task.cancel() + if manager._sweeper_task is not None and not manager._sweeper_task.done(): + manager._sweeper_task.cancel() + + +def make_ctx(**overrides): + defaults = dict( + session_id="s1", + event_sink=None, + stop_event=None, + model="m", + user_id="u1", + user_role="admin", + user_info=None, + cwd="/tmp", + ) + defaults.update(overrides) + return ToolContext(**defaults) + + +class TestLifecycle: + async def test_completed_job(self, fresh_manager): + async def factory(bg_ctx): + return ToolResult(success=True, output="done 42") + + job = fresh_manager.submit("s1", "code_exec", {"code": "x"}, factory, make_ctx()) + assert job.status == "running" + await job.done.wait() + assert job.status == "completed" + assert job.result.output == "done 42" + assert job.finished_at is not None + + async def test_failed_job(self, fresh_manager): + async def factory(bg_ctx): + raise ValueError("boom") + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await job.done.wait() + assert job.status == "failed" + assert "boom" in job.error + assert job.result.success is False + + async def test_cancel_running(self, fresh_manager): + started = asyncio.Event() + + async def factory(bg_ctx): + started.set() + await asyncio.sleep(30) + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await started.wait() + assert fresh_manager.cancel(job) is True + await job.done.wait() + assert job.status == "cancelled" + assert fresh_manager.cancel(job) is False # already finished + + async def test_task_stop_event_is_isolated(self, fresh_manager): + """The job sees its own stop_event, not any run-level one.""" + seen = {} + + async def factory(bg_ctx): + seen["ctx_stop"] = bg_ctx.stop_event + seen["var_stop"] = current_stop_event.get() + return ToolResult(success=True, output="ok") + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await job.done.wait() + assert seen["ctx_stop"] is job.stop_event + assert seen["var_stop"] is job.stop_event + + async def test_get_is_session_scoped(self, fresh_manager): + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await job.done.wait() + assert fresh_manager.get(job.task_id, "s1") is job + assert fresh_manager.get(job.task_id, "s2") is None + assert [j.task_id for j in fresh_manager.list("s1")] == [job.task_id] + assert fresh_manager.list("s2") == [] + + +class TestCaps: + async def test_per_session_cap(self, fresh_manager, monkeypatch): + patch_settings(monkeypatch, tasks_max_per_session=1) + + async def factory(bg_ctx): + await asyncio.sleep(30) + + first = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert not isinstance(first, str) + second = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert isinstance(second, str) and "limit" in second + fresh_manager.cancel(first) + + async def test_rate_limit(self, fresh_manager, monkeypatch): + patch_settings(monkeypatch, tasks_rate_limit=1) + + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + first = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert not isinstance(first, str) + second = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert isinstance(second, str) and "rate limit" in second + + async def test_spawn_cap_only_limits_spawn_agent(self, fresh_manager, monkeypatch): + patch_settings(monkeypatch, tasks_max_spawn=1) + + async def factory(bg_ctx): + await asyncio.sleep(30) + + first = fresh_manager.submit("s1", "spawn_agent", {}, factory, make_ctx()) + assert not isinstance(first, str) + # another spawn_agent is rejected… + second = fresh_manager.submit("s1", "spawn_agent", {}, factory, make_ctx()) + assert isinstance(second, str) and "subagent" in second + # …but a regular tool is not + third = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert not isinstance(third, str) + fresh_manager.cancel(first) + fresh_manager.cancel(third) + + +class TestUpdateCallback: + async def test_published_on_submit_and_finish(self, fresh_manager): + updates = [] + fresh_manager.set_update_callback(updates.append) + + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + job = fresh_manager.submit("s1", "terminal", {"command": "ls"}, factory, make_ctx(), + parent_tool_call_id="tc1") + await job.done.wait() + await asyncio.sleep(0) # let the callback fire settle + statuses = [u.status for u in updates] + assert statuses[0] == "running" + assert statuses[-1] == "completed" + final = updates[-1] + assert isinstance(final, TaskUpdate) + assert final.task_id == job.task_id + assert final.tool == "terminal" + assert final.parent_tool_call_id == "tc1" + assert final.to_wire()["type"] == "task_update" + + async def test_broken_callback_does_not_kill_job(self, fresh_manager): + def bad_callback(update): + raise RuntimeError("nope") + + fresh_manager.set_update_callback(bad_callback) + + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await job.done.wait() + assert job.status == "completed" + + +class TestBoundedEventQueue: + async def test_drop_oldest_when_full(self): + q = BoundedEventQueue(maxsize=2) + await q.put("a") + q.put_nowait("b") + await q.put("c") + assert list(q._queue) == ["b", "c"] + + async def test_put_never_blocks(self): + q = BoundedEventQueue(maxsize=1) + await asyncio.wait_for(q.put("a"), timeout=0.1) + await asyncio.wait_for(q.put("b"), timeout=0.1) + assert q.qsize() == 1 + + +class TestReap: + async def test_reap_drops_expired_finished_jobs(self, fresh_manager, monkeypatch): + patch_settings(monkeypatch, tasks_ttl_sec=0) + + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + await job.done.wait() + assert fresh_manager.reap() == 1 + assert fresh_manager.get(job.task_id, "s1") is None + + async def test_reap_keeps_running_jobs(self, fresh_manager, monkeypatch): + patch_settings(monkeypatch, tasks_ttl_sec=0) + + async def factory(bg_ctx): + await asyncio.sleep(30) + + job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert fresh_manager.reap() == 0 + assert fresh_manager.get(job.task_id, "s1") is job + fresh_manager.cancel(job) + + +class TestHelpers: + def test_get_task_manager_singleton(self, monkeypatch): + monkeypatch.setattr(tasks_mod, "_manager", None) + assert get_task_manager() is get_task_manager() + + def test_args_summary(self): + assert args_summary({"a": 1}) == '{"a": 1}' + assert args_summary({"a": "x" * 500}, limit=10) == '{"a": "xx' + "x" + + async def test_task_job_defaults(self): + job = TaskJob(task_id="bt-x", session_id="s1", tool="terminal", args={}) + assert job.status == "running" + assert job.subagent_tokens is None + assert job.preview() == "" # running → empty preview + + async def test_time_in_submit_window(self, fresh_manager): + async def factory(bg_ctx): + return ToolResult(success=True, output="ok") + + fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx()) + assert len(fresh_manager._submit_times["s1"]) == 1 + assert time.time() - fresh_manager._submit_times["s1"][0] < 5 \ No newline at end of file diff --git a/tests/unit/core/test_task_notes.py b/tests/unit/core/test_task_notes.py new file mode 100644 index 0000000..b06491b --- /dev/null +++ b/tests/unit/core/test_task_notes.py @@ -0,0 +1,114 @@ +"""Unit tests for navi.core.task_notes — pending background-task completion notes.""" + +import pytest + +from navi.core import task_notes + + +class FakeKv: + """In-memory stand-in for KvStore.get/set.""" + + def __init__(self): + self.data = {} + + async def get(self, user_id, session_id, scope, key): + return self.data.get((user_id, session_id, scope, key)) + + async def set(self, user_id, session_id, scope, key, value): + self.data[(user_id, session_id, scope, key)] = value + + +def patch_settings(monkeypatch, **overrides): + """Replace the frozen settings singleton with an overridden copy.""" + import navi.config as config_mod + from navi.config import Settings + + new_settings = Settings(**overrides) + monkeypatch.setattr(config_mod, "settings", new_settings) + return new_settings + + +@pytest.fixture(autouse=True) +def kv(monkeypatch): + store = FakeKv() + task_notes.set_kv_store(store) + yield store + task_notes.set_kv_store(None) + + +def make_job(task_id="bt-ab12", session_id="s1", status="completed", + preview_text="done 42", tokens=None): + class Job: + pass + + job = Job() + job.task_id = task_id + job.session_id = session_id + job.tool = "terminal" + job.status = status + job.subagent_tokens = tokens + job.preview = lambda limit=800: preview_text + return job + + +class TestAddNote: + async def test_note_recorded(self, kv): + await task_notes.add_note(make_job()) + assert await task_notes.pending_count("s1") == 1 + + async def test_cap_drops_oldest(self, kv, monkeypatch): + patch_settings(monkeypatch, task_notes_max_pending=2) + for i in range(4): + await task_notes.add_note(make_job(task_id=f"bt-{i}")) + assert await task_notes.pending_count("s1") == 2 + notes = await task_notes._load("s1") + assert [n["task_id"] for n in notes] == ["bt-2", "bt-3"] + + async def test_no_store_is_noop(self, monkeypatch): + task_notes.set_kv_store(None) + await task_notes.add_note(make_job()) # must not raise + assert await task_notes.drain("s1") is None + + +class TestDrain: + async def test_drain_coalesces_and_empties(self, kv): + await task_notes.add_note(make_job(task_id="bt-a", preview_text="42 files")) + await task_notes.add_note(make_job(task_id="bt-b", status="failed", + preview_text="boom")) + text = await task_notes.drain("s1") + assert text is not None + assert "[Background task results]" in text + assert "bt-a (terminal) completed: 42 files" in text + assert "bt-b (terminal) failed: boom" in text + assert "Do not start new background tasks" in text + # queue is now empty + assert await task_notes.drain("s1") is None + assert await task_notes.pending_count("s1") == 0 + + async def test_drain_limits_per_turn(self, kv, monkeypatch): + patch_settings(monkeypatch, task_notes_per_turn=2) + for i in range(3): + await task_notes.add_note(make_job(task_id=f"bt-{i}")) + text = await task_notes.drain("s1") + assert "bt-0" in text and "bt-1" in text + assert "bt-2" not in text.split("older result")[0] + assert "1 older result(s)" in text + # remaining note survives for the next turn + assert await task_notes.pending_count("s1") == 1 + text2 = await task_notes.drain("s1") + assert "bt-2" in text2 + + async def test_drain_scoped_by_session(self, kv): + await task_notes.add_note(make_job(session_id="s1")) + assert await task_notes.drain("s2") is None + assert await task_notes.drain("s1") is not None + + async def test_empty_preview_placeholder(self, kv): + await task_notes.add_note(make_job(preview_text="")) + text = await task_notes.drain("s1") + assert "(no output)" in text + + async def test_subagent_tokens_recorded(self, kv): + await task_notes.add_note(make_job(tokens=1234)) + notes = await task_notes._load("s1") + assert notes[0]["subagent_tokens"] == 1234 \ No newline at end of file diff --git a/tests/unit/core/test_task_notes_injection.py b/tests/unit/core/test_task_notes_injection.py new file mode 100644 index 0000000..8ab78e7 --- /dev/null +++ b/tests/unit/core/test_task_notes_injection.py @@ -0,0 +1,89 @@ +"""Unit tests for background-note injection into the session (Agent._drain_task_notes).""" + +from types import SimpleNamespace + +import pytest + +from navi.core import task_notes +from navi.core.agent import Agent + + +class FakeKv: + def __init__(self): + self.data = {} + + async def get(self, user_id, session_id, scope, key): + return self.data.get((user_id, session_id, scope, key)) + + async def set(self, user_id, session_id, scope, key, value): + self.data[(user_id, session_id, scope, key)] = value + + +class FakeStore: + def __init__(self): + self.saved = [] + + async def save(self, session): + self.saved.append(session) + + +@pytest.fixture(autouse=True) +def kv(monkeypatch): + store = FakeKv() + task_notes.set_kv_store(store) + yield store + task_notes.set_kv_store(None) + + +def make_agent_with_store(): + agent = object.__new__(Agent) + agent._sessions = FakeStore() + return agent + + +def make_session(): + return SimpleNamespace(context=[], messages=[]) + + +class TestDrainTaskNotes: + async def test_note_injected_as_system_message_and_saved(self): + agent = make_agent_with_store() + session = make_session() + job = SimpleNamespace( + task_id="bt-ab12", session_id="s1", tool="terminal", status="completed", + subagent_tokens=None, preview=lambda limit=800: "done 42", + ) + await task_notes.add_note(job) + + await agent._drain_task_notes("s1", session) + + assert len(session.context) == 1 + note = session.context[0] + assert note.role == "system" + assert "[Background task results]" in note.content + assert "bt-ab12" in note.content + assert note.metadata.get("source") == "task_note" + # note did NOT go into the display history + assert session.messages == [] + # session was persisted + assert agent._sessions.saved == [session] + + async def test_no_notes_nothing_injected(self): + agent = make_agent_with_store() + session = make_session() + + await agent._drain_task_notes("s1", session) + + assert session.context == [] + assert agent._sessions.saved == [] + + async def test_drain_failure_is_swallowed(self, monkeypatch): + agent = make_agent_with_store() + session = make_session() + + async def broken_drain(session_id): + raise RuntimeError("kv down") + + monkeypatch.setattr(task_notes, "drain", broken_drain) + await agent._drain_task_notes("s1", session) # must not raise + assert session.context == [] \ No newline at end of file diff --git a/tests/unit/core/test_tool_executor.py b/tests/unit/core/test_tool_executor.py index 3888514..26eec7d 100644 --- a/tests/unit/core/test_tool_executor.py +++ b/tests/unit/core/test_tool_executor.py @@ -1,11 +1,47 @@ """Unit tests for navi.core.tool_executor.""" +import asyncio +import json + +import pytest + +from navi.core import tasks as tasks_mod from navi.core.registry import ToolRegistry from navi.core.tool_executor import ToolExecutor -from navi.llm.base import ToolCallRequest +from navi.llm.base import Message, ToolCallRequest +from navi.tools._internal.base import ToolResult from tests.conftest_factory import FakeTool +def patch_settings(monkeypatch, **overrides): + import navi.config as config_mod + from navi.config import Settings + + new_settings = Settings(**overrides) + monkeypatch.setattr(config_mod, "settings", new_settings) + return new_settings + + +class _Ctx: + """Minimal ToolContext stand-in carrying only the session id.""" + + def __init__(self, session_id: str = "s1") -> None: + self.session_id = session_id + + +class RecordingTool: + """Fake tool that captures its arguments before answering.""" + + def __init__(self, name: str, output: str = "ok") -> None: + self.name = name + self.output = output + self.calls: list[dict] = [] + + async def execute(self, arguments: dict, ctx=None) -> ToolResult: + self.calls.append(dict(arguments)) + return ToolResult(success=True, output=self.output) + + class TestToolExecutorMcpAliases: async def test_executes_bare_mcp_tool_alias(self): registry = ToolRegistry() @@ -66,3 +102,148 @@ assert images == [] assert messages[0].name == "mcp__gnexus_book__search_docs" assert messages[0].content == "executed mcp__gnexus_book__search_docs" + + +class TestBackgroundInterception: + @pytest.fixture(autouse=True) + def _fresh_manager(self, monkeypatch): + from navi.core.tasks import TaskManager + + manager = TaskManager() + monkeypatch.setattr(tasks_mod, "_manager", manager) + self.manager = manager + yield manager + for job in list(manager._jobs.values()): + if job.task is not None and not job.task.done(): + job.task.cancel() + + async def test_background_flag_detaches_call(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="slow_tool") + tool = RecordingTool("slow_tool", output="long result") + registry = ToolRegistry() + registry.register(tool, builtin=True) + executor = ToolExecutor(registry) + + event, msg, image = await executor._execute_one( + ToolCallRequest(id="tc1", name="slow_tool", + arguments={"query": "x", "background": True}), + {"slow_tool": tool}, + ctx=_Ctx(), + ) + + # immediate synthetic result with a task id + assert msg.metadata.get("background") is True + payload = json.loads(msg.content) + task_id = payload["task_id"] + assert task_id.startswith("bt-") + assert payload["status"] == "running" + assert "tasks check" in payload["hint"] + assert event.success is True + # background flag is stripped from the args passed to the tool + job = self.manager.get(task_id, "s1") + assert job is not None + assert "background" not in job.args + await job.done.wait() + assert tool.calls == [{"query": "x"}] + assert job.result.output == "long result" + + async def test_non_backgroundable_tool_runs_inline(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="other_tool") + tool = RecordingTool("slow_tool") + registry = ToolRegistry() + registry.register(tool, builtin=True) + executor = ToolExecutor(registry) + + event, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="slow_tool", + arguments={"query": "x", "background": True}), + {"slow_tool": tool}, + ctx=_Ctx(), + ) + + assert msg.content == "ok" # ran inline, real output + assert msg.metadata.get("background") is None + assert tool.calls == [{"query": "x", "background": True}] # flag intact + + async def test_background_false_runs_inline(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="slow_tool") + tool = RecordingTool("slow_tool") + executor = ToolExecutor({"slow_tool": tool}) + + _, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="slow_tool", + arguments={"background": False}), + {"slow_tool": tool}, + ctx=_Ctx(), + ) + assert msg.content == "ok" + assert self.manager.list("s1") == [] + + async def test_cap_rejection_is_inline_failure(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="slow_tool", + tasks_max_per_session=0) + tool = RecordingTool("slow_tool") + executor = ToolExecutor({"slow_tool": tool}) + + _, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="slow_tool", + arguments={"background": True}), + {"slow_tool": tool}, + ctx=_Ctx(), + ) + assert "Cannot run in background" in msg.content + assert msg.metadata.get("background") is None + assert tool.calls == [] # was not executed + + async def test_terminal_open_not_hijacked(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="terminal") + tool = RecordingTool("terminal") + executor = ToolExecutor({"terminal": tool}) + + _, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="terminal", + arguments={"action": "open", "background": True}), + {"terminal": tool}, + ctx=_Ctx(), + ) + # native terminal open(background=true) semantics preserved + assert msg.content == "ok" + assert tool.calls == [{"action": "open", "background": True}] + assert self.manager.list("s1") == [] + + async def test_terminal_run_is_backgroundable(self, monkeypatch): + patch_settings(monkeypatch, backgroundable_tools="terminal") + tool = RecordingTool("terminal") + executor = ToolExecutor({"terminal": tool}) + + _, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="terminal", + arguments={"action": "run", "command": "sleep 5", + "background": True}), + {"terminal": tool}, + ctx=_Ctx(), + ) + assert msg.metadata.get("background") is True + job = list(self.manager.list("s1"))[0] + assert job.args == {"action": "run", "command": "sleep 5"} + await job.done.wait() + + async def test_spawn_agent_gets_deep_ring(self, monkeypatch): + """Ф2: detached sub-agents get a 200-event ring for live progress.""" + from types import SimpleNamespace + + patch_settings(monkeypatch, backgroundable_tools="spawn_agent") + tool = RecordingTool("spawn_agent") + executor = ToolExecutor({"spawn_agent": tool}) + + _, msg, _ = await executor._execute_one( + ToolCallRequest(id="tc1", name="spawn_agent", + arguments={"task": "research", "background": True}), + {"spawn_agent": tool}, + ctx=SimpleNamespace(session_id="s1"), + ) + task_id = json.loads(msg.content)["task_id"] + job = self.manager.get(task_id, "s1") + assert job.ring.maxsize == 200 + assert job.parent_tool_call_id == "tc1" + await job.done.wait() diff --git a/tests/unit/test_deployment_flags.py b/tests/unit/test_deployment_flags.py index 819be2a..a69820b 100644 --- a/tests/unit/test_deployment_flags.py +++ b/tests/unit/test_deployment_flags.py @@ -17,7 +17,24 @@ ROUTES_SCRIPT = """ import json import navi.main as m -print(json.dumps(sorted({r.path for r in m.app.routes}))) + +# Newer Starlette nests include_router results as _IncludedRouter objects +# without a top-level .path — walk nested routes recursively. +def walk(routes, out): + for r in routes: + p = getattr(r, "path", None) + if p: + out.add(p) + nested = getattr(r, "routes", None) + if nested is None: + original = getattr(r, "original_router", None) + nested = getattr(original, "routes", None) if original else None + if nested: + walk(nested, out) + +out = set() +walk(m.app.routes, out) +print(json.dumps(sorted(out))) """ diff --git a/tests/unit/tools/test_spawn_agent.py b/tests/unit/tools/test_spawn_agent.py index b542312..9537be8 100644 --- a/tests/unit/tools/test_spawn_agent.py +++ b/tests/unit/tools/test_spawn_agent.py @@ -73,3 +73,14 @@ assert result.success is False assert result.error == "unknown_profile:missing_profile" assert "Available profiles" in result.output + + +@pytest.mark.anyio +async def test_schema_declares_background_param(spawn_tool): + """Ф2: the tool schema exposes background=true for detached sub-agents.""" + tool, _, _ = spawn_tool + background = tool.parameters["properties"]["background"] + assert background["type"] == "boolean" + assert "task_id" in background["description"] + # the default remains synchronous + assert "SYNCHRONOUS" in tool.description diff --git a/tests/unit/tools/test_tasks_tool.py b/tests/unit/tools/test_tasks_tool.py new file mode 100644 index 0000000..408e38d --- /dev/null +++ b/tests/unit/tools/test_tasks_tool.py @@ -0,0 +1,155 @@ +"""Unit tests for the tasks tool (list/check/wait/cancel of background jobs).""" + +import asyncio + +import pytest + +from navi.core import tasks as tasks_mod +from navi.core.tasks import TaskManager +from navi.tools._internal.base import ToolContext, ToolResult +from navi.tools.tasks import TasksTool + + +@pytest.fixture(autouse=True) +def fresh_manager(monkeypatch): + manager = TaskManager() + monkeypatch.setattr(tasks_mod, "_manager", manager) + yield manager + for job in list(manager._jobs.values()): + if job.task is not None and not job.task.done(): + job.task.cancel() + if manager._sweeper_task is not None and not manager._sweeper_task.done(): + manager._sweeper_task.cancel() + + +def make_ctx(session_id="s1"): + return ToolContext( + session_id=session_id, event_sink=None, stop_event=None, model=None, + user_id=None, user_role="admin", user_info=None, cwd=None, + ) + + +async def submit_ok(manager, session_id="s1", output="done", delay=0.0): + async def factory(bg_ctx): + if delay: + await asyncio.sleep(delay) + return ToolResult(success=True, output=output) + + return manager.submit(session_id, "code_exec", {"code": "x"}, factory, + make_ctx(session_id)) + + +class TestList: + async def test_empty(self): + result = await TasksTool().execute({"action": "list"}, ctx=make_ctx()) + assert result.success + assert "No background tasks" in result.output + + async def test_lists_session_jobs_only(self, fresh_manager): + await submit_ok(fresh_manager, "s1") + await submit_ok(fresh_manager, "s2") + result = await TasksTool().execute({"action": "list"}, ctx=make_ctx("s1")) + assert result.success + assert "code_exec" in result.output + + async def test_list_shows_finished_preview(self, fresh_manager): + job = await submit_ok(fresh_manager, "s1", output="42 files") + await job.done.wait() + result = await TasksTool().execute({"action": "list"}, ctx=make_ctx()) + assert "completed" in result.output + assert "42 files" in result.output + + +class TestCheck: + async def test_check_finished_shows_result(self, fresh_manager): + job = await submit_ok(fresh_manager, output="the answer") + await job.done.wait() + result = await TasksTool().execute( + {"action": "check", "task_id": job.task_id}, ctx=make_ctx()) + assert result.success + assert "completed" in result.output + assert "the answer" in result.output + + async def test_check_running_shows_progress_placeholder(self, fresh_manager): + job = await submit_ok(fresh_manager, delay=5) + result = await TasksTool().execute( + {"action": "check", "task_id": job.task_id}, ctx=make_ctx()) + assert "running" in result.output + fresh_manager.cancel(job) + + async def test_check_unknown_task(self): + result = await TasksTool().execute( + {"action": "check", "task_id": "bt-nope"}, ctx=make_ctx()) + assert not result.success + assert result.error == "task_not_found" + + async def test_task_id_required(self): + result = await TasksTool().execute({"action": "check"}, ctx=make_ctx()) + assert not result.success + + async def test_check_is_session_scoped(self, fresh_manager): + job = await submit_ok(fresh_manager, session_id="other") + result = await TasksTool().execute( + {"action": "check", "task_id": job.task_id}, ctx=make_ctx("s1")) + assert result.error == "task_not_found" + + +class TestWait: + async def test_wait_returns_result(self, fresh_manager): + job = await submit_ok(fresh_manager, output="late result", delay=0.05) + result = await TasksTool().execute( + {"action": "wait", "task_id": job.task_id}, ctx=make_ctx()) + assert result.success + assert "late result" in result.output + + async def test_wait_timeout(self, fresh_manager): + job = await submit_ok(fresh_manager, delay=30) + result = await TasksTool().execute( + {"action": "wait", "task_id": job.task_id, "timeout": 0.05}, + ctx=make_ctx()) + assert not result.success + assert result.error == "wait_timeout" + assert "still running" in result.output + fresh_manager.cancel(job) + + async def test_wait_timeout_capped_at_120(self, fresh_manager): + job = await submit_ok(fresh_manager, delay=0.01) + # timeout above the cap must not blow up; job finishes fast anyway + result = await TasksTool().execute( + {"action": "wait", "task_id": job.task_id, "timeout": 9999}, + ctx=make_ctx()) + assert result.success + + +class TestCancel: + async def test_cancel_running(self, fresh_manager): + job = await submit_ok(fresh_manager, delay=30) + result = await TasksTool().execute( + {"action": "cancel", "task_id": job.task_id}, ctx=make_ctx()) + assert result.success + await job.done.wait() + assert job.status == "cancelled" + + async def test_cancel_finished(self, fresh_manager): + job = await submit_ok(fresh_manager) + await job.done.wait() + result = await TasksTool().execute( + {"action": "cancel", "task_id": job.task_id}, ctx=make_ctx()) + assert not result.success + assert result.error == "not_running" + + +class TestFallbacks: + async def test_session_falls_back_to_contextvar(self, fresh_manager, monkeypatch): + from navi.tools._internal.base import current_session_id + + job = await submit_ok(fresh_manager, "ctx-session") + token = current_session_id.set("ctx-session") + try: + await asyncio.sleep(0) # let the background job run + result = await TasksTool().execute( + {"action": "check", "task_id": job.task_id}, ctx=None) + finally: + current_session_id.reset(token) + assert result.success + assert "completed" in result.output \ No newline at end of file