From c436251be4b7fa5ce57bb8ef5899dd59259eed0a Mon Sep 17 00:00:00 2001 From: sebi Date: Tue, 1 Sep 2026 16:10:05 -0500 Subject: [PATCH 1/2] feat(sdk): upgrade to claude-agent-sdk 0.2.150 and adopt what it changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Thirteen releases; twelve are bundled-CLI bumps (Claude Code 2.1.229 → 2.1.257). The one substantive release is 0.2.140, and three of its four features land here. Pin mcp <2, because the upgrade breaks claudetm-mcp without it. mcp 2.x renamed FastMCP to MCPServer and removed mcp.server.fastmcp, which every module under claude_task_master/mcp/ imports. Our extra was an unbounded mcp>=1.26.0; until 0.2.140 the SDK's own mcp<2.0.0 pin was silently holding the line, and it widened to mcp<3.0.0. Verified: installing claude-task-master[mcp] resolves mcp 2.1.1 today and the import raises. Classify errors from the payload, not the prose (agent_error_classify). The verdict decides whether an unattended run retries or dies, and it was substring matching on str(error). ResultError now carries api_error_status/subtype/terminal_reason/errors/result, so the structured payload is consulted first. str(ResultError) names only the subtype, so the payload is folded into the searched text too — which is what makes 529 overloaded a retryable APIServerError and lets "Connection closed mid-response" classify at all. Two loose rules are fixed with token boundaries: "500" also matched "took 1500ms", and "auth" also matched "Co-Authored-By", which appears in every commit message we write. Forward worker text (forward_subagent_text). The SDK forwards a subagent's tool calls unasked but withheld its prose, leaving a fanned-out session half-visible. The rendering already existed. Hive ceiling 10 → 6, and enforced rather than merely stated. The number is interpolated into the fan-out brief, so it is prompt-visible. The CLI hands out concurrency slots and refuses the overflow, reading CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS (default 20), so we pass ours. The brief also now says the composition is the lead's: N of the same kind, N different kinds, or any mix — nothing rewards variety. Add a live roster for fanned-out sessions (hive_roster). Per-worker prefixes say who spoke; they cannot say how many workers are live, what each is on, and what each has burned. Two stream facts, both established by probing a live session rather than by reading, decide how it is fed: a dispatch's tool result is an ACK arriving ~0.1s later while the worker runs on for another 40s (the first cut rendered "3 done" at 0s with all three still working), and the roster adopts unfamiliar ids by design, so completion is gated on a recorded dispatch. Regenerate requirements.txt, stale at claude-agent-sdk==0.1.35. Refs #160, #161 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Phc3xittodp1pz7ZeMRAfa --- CHANGELOG.md | 35 + CLAUDE.md | 30 +- README.md | 6 +- pyproject.toml | 11 +- requirements.txt | 19 +- .../core/agent_error_classify.py | 233 +++++++ src/claude_task_master/core/agent_message.py | 48 +- .../core/agent_message_roster.py | 98 +++ .../core/agent_query_execute.py | 26 + .../core/agent_query_helpers.py | 71 +- src/claude_task_master/core/hive.py | 65 +- src/claude_task_master/core/hive_roster.py | 499 ++++++++++++++ .../core/prompts_working_hive.py | 9 +- tests/core/test_agent_error_classify.py | 523 ++++++++++++++ tests/core/test_agent_message.py | 116 ++++ tests/core/test_hive.py | 55 +- tests/core/test_hive_roster.py | 640 ++++++++++++++++++ uv.lock | 19 +- 18 files changed, 2420 insertions(+), 83 deletions(-) create mode 100644 src/claude_task_master/core/agent_error_classify.py create mode 100644 src/claude_task_master/core/agent_message_roster.py create mode 100644 src/claude_task_master/core/hive_roster.py create mode 100644 tests/core/test_agent_error_classify.py create mode 100644 tests/core/test_hive_roster.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b4fbd589..26c85726 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,41 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **Upgraded `claude-agent-sdk` to `>=0.2.150,<0.3`** (from `0.2.137`). Thirteen releases; twelve of them are bundled-CLI bumps carrying Claude Code **2.1.229 → 2.1.257**. The one substantive release is 0.2.140, and three of its four features land here (`ResultError`, `forward_subagent_text`, and the MCP 2.x widening that forced the pin below); `can_use_tool` for string prompts is not used — claudetm runs `permission_mode="bypassPermissions"` and has no permission callback to install. +- **`CLAUDETM_HIVE_MAX_PARALLEL` default 10 → 6.** This constant is **prompt-visible**: it is interpolated verbatim into the fan-out brief as the ceiling a lead may dispatch up to, so it is not an internal guard that can be tuned quietly. Ten invited a split wider than a single shared checkout on one box rewards — every worker is a full agent process re-reading the repo on the same cores, and past a handful the cold starts and the lead's own verification pass cost more than the concurrency returns. Still a ceiling and never a target; zero remains the right answer for most tasks. +- **The brief now states that the team's *composition* is the lead's, and that nothing rewards variety.** Up to the ceiling, workers may be N of the same kind, one each of N different kinds, or any mix — four concurrent `hive-worker`s is exactly as legitimate as a `backend-dev`, a `frontend-dev` and two `hive-worker`s. What selects a worker is the piece it is handed: the specialist whose description fits it, else generic `hive-worker`. Two pieces suiting the same specialist get two instances of it. Previously the brief named the ceiling and the specialist-first rule but never said whether repeating a type was allowed, which reads as if it is not. +- **Regenerated `requirements.txt`**, a committed `uv pip compile` artifact that had gone stale at `claude-agent-sdk==0.1.35` — roughly 115 releases behind, and unusable against this codebase. Nothing in CI or the Dockerfile consumes it, but claudetm's own repo-setup handler runs `uv pip install -r requirements.txt` whenever it finds one, so the file was a live trap for anyone it was pointed at. + +### Added + +- **A fanned-out session now shows its team, not just its lines** (`core/hive_roster.py`, `CLAUDETM_HIVE_ROSTER_INTERVAL_SEC`, default 60s, `0` disables). The per-worker `↳ [hive-worker#2]` prefixes answer "who said this" and cannot answer the questions you actually have while watching a hive: how many workers are live, what each is on right now, how long each has been going, and what each has burned. The roster tracks every worker from its dispatch block to its returning tool result and renders a compact block — forced when the team's shape changes, throttled otherwise: + + ``` + Hive: 2 workers active, 1 done + ~ hive-worker#1 Bash uv run pytest tests/core/test_auth.py 5m 11s in 410.2k out 8.1k + ~ backend-dev#2 Edit src/api/routes.py 5m 11s in 180.2k out 6.0k + + hive-worker#3 done 5m 11s in 41.0k out 2.2k + ``` + + Two stream facts, both established by probing a live session rather than by reading, decide how this is fed. **A dispatch's tool result is an acknowledgement, not a completion**: the `ToolResultBlock` for an `Agent` call arrives ~0.1s after the dispatch while that worker's own messages keep arriving for the next 40 seconds — the first cut read it as "returned" and rendered `3 done` at `0s` elapsed with all three still working. Only a *failed* dispatch now retires a worker. And the roster **adopts an unfamiliar tool-use id** by design (a worker can speak before its spawning block is processed), so completion is additionally gated on an id already recorded as a dispatch — otherwise every failed ordinary Read/Bash call invents a phantom worker on sessions with no hive at all. + + Per-worker tokens come from `AssistantMessage.usage` on messages carrying `parent_tool_use_id` — the only per-subagent figure the stream offers, since the terminal `ResultMessage` aggregates the lead and every worker into the one total that cannot say *which* worker spent it. The SDK does not document that field's exact meaning, so it is best-effort: malformed values are ignored rather than reported as a number. + +- **The hive ceiling is enforced by the runtime, not just stated in the prompt.** `CLAUDETM_HIVE_MAX_PARALLEL` had always been prose — a number interpolated into the brief with nothing behind it — and this codebase has already measured what a prose-only rule is worth (leads ignored the "never background a worker" instruction in 27% of dispatches, which is why that one is pinned on the agent definition). The bundled CLI enforces a cap of its own: it hands out concurrency slots and refuses the overflow dispatch with `Concurrent subagent limit reached`, reading `CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS` and defaulting to 20 when unset. claudetm now passes its own ceiling there, so the brief and the runtime agree on one number. + +- **Workers' own prose reaches the session log, not just their tool calls** (`forward_subagent_text`, new in SDK 0.2.140; `CLAUDETM_FORWARD_SUBAGENT_TEXT`, default on). The SDK forwards a subagent's tool calls unasked but withheld its text and thinking, which left a fanned-out session half-visible: you could see *that* a worker edited a file and never what it was trying to do — on a run where the worker's final message is the only thing that comes back to the lead, and the log is the only forensic record. The rendering it needs already existed: stable per-worker colour and `#n` ordinal, and subagent text displayed but never accumulated into the lead's own result. Forwarding is display-only and costs no tokens — those blocks were generated either way; the only price is log volume. Applied solely to sessions that may fan out, since a session denied the dispatch tools has no subagent to forward. + +### Fixed + +- **`mcp` is now pinned `<2`, because upgrading the SDK breaks `claudetm-mcp` without it.** mcp 2.x renamed `FastMCP` to `MCPServer` and removed `mcp.server.fastmcp`, which every module under `claude_task_master/mcp/` imports. claudetm's extra was the unbounded `mcp>=1.26.0`; until 0.2.140 the SDK's own `mcp<2.0.0` pin was silently holding the line for us, and 0.2.140 widened it to `mcp<3.0.0`. Verified: `pip install "claude-task-master[mcp]==0.1.89"` resolves **mcp 2.1.1** today, and `from mcp.server.fastmcp import FastMCP` then raises `ModuleNotFoundError`. Lift the pin only together with the MCPServer migration. +- **Error classification read the prose and could not see the payload** (`core/agent_error_classify.py`, extracted from `agent_query_helpers`). The verdict decides whether an unattended run **retries or dies** — `TRANSIENT_ERRORS` members retry under the failure budget, everything else propagates — and it was derived from substring matching on `str(error)` alone. SDK 0.2.140 raises `ResultError` carrying `api_error_status`, `subtype`, `terminal_reason`, `errors` and `result`, so the structured payload is now consulted first: 429 → rate limit, 401/403 → auth, 404 *with* a model mentioned → the fallback chain, 408/504 → timeout, any other 5xx → server error. An HTTP status is a fact; a substring is a guess. + + `str(ResultError)` is **only** `"Claude Code returned an error result: (exit code: 1)"` — the prose naming the actual failure lives in `errors`/`result`. Folding the payload into the searched text is what makes **529 overloaded** a retryable `APIServerError` (it appears nowhere in `str()`, so the old rules classified it as an unrecognised error) and what lets `Connection closed mid-response` — the blip that once ended a 22-task unattended run at task 1 — classify as `APIConnectionError` at all. + +- **Two loose substring rules, each wrong in one direction.** `"500" in text` also matched `request took 1500ms`, promoting a latency message to a retryable `APIServerError`; `"auth" in text` also matched **`Co-Authored-By`**, which appears in every commit message this project writes, so any error echoing a git command became a *non*-retryable `APIAuthenticationError` that ended the run. Status codes and auth phrasings now match on token boundaries, pinned by named regression tests. + ## [0.1.89] - 2026-08-14 ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index 7f9c99e9..9c8f8177 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -233,6 +233,30 @@ Unfinished → the task is **not** checked off and the same task re-runs, with a Relatedly, an unclassified `QueryExecutionError` (a CLI crash whose text carries no keyword `_classify_api_error` recognises) is no longer fatal on sight: it retries under the same failure budget as a connection error (`rate_limit_config.max_retries`), and only a persistent one raises `ConsecutiveFailuresError`. +### Error classification reads the payload, not the prose (`core/agent_error_classify.py`) + +The verdict decides whether an unattended run **retries or dies** — `TRANSIENT_ERRORS` members are +retried under the failure budget, everything else propagates — so it is no longer derived from +substring matching alone. `claude-agent-sdk` >= 0.2.140 raises **`ResultError`** (a `ProcessError` +subclass) when the CLI exits after a terminal error result, carrying `api_error_status`, `subtype`, +`terminal_reason`, `errors` and `result`. An HTTP status is a fact; a substring is a guess, so the +structured payload is consulted first: 429 → rate limit, 401/403 → auth, 404 *with* a model +mentioned → the fallback chain, 408/504 → timeout, any other 5xx → server error. + +`str(ResultError)` is **only** `"Claude Code returned an error result: (exit code: 1)"` — +the prose naming the actual failure lives in `errors`/`result`. So the payload is folded into the +searched text as well, which is what lets `Connection closed mid-response` (the blip documented +above as ending a 22-task run) classify as a retryable `APIConnectionError` at all, and what makes +**529 overloaded** — invisible to the old rules, absent from `str()` — a retryable `APIServerError`. + +The text rules now match on **token boundaries**. Two one-way misclassifications lived in the loose +version: `"500" in text` also matched `request took 1500ms`, promoting a latency message to a +retryable server error; and `"auth" in text` also matched **`Co-Authored-By`**, which appears in +every commit message this project writes, so any error echoing a git command became a +*non*-retryable `APIAuthenticationError` that ended the run. Both are pinned by named regression +tests. The module reads the SDK fields by `getattr` and never imports `claude_agent_sdk` — it must +stay importable in exactly the situation some of these errors describe. + ### Accumulated context is capped where it is read `context.md` grows by a summary per session and never shrinks, so what enters a prompt has to be @@ -345,7 +369,11 @@ A triplicated `plan.md` still runs its work N times; the parser only guarantees - **The seam is the API, not the path.** Disjoint file sets are necessary and not sufficient: a piece that renames an export, changes a signature or edits a shared type breaks every file importing it, *including files no worker owns*. The lead lists, per piece, what it changes that something else reads, and either pulls every caller into that set or keeps the piece itself. (Observed live: one worker renamed a field across the files it owned, an unowned file kept the old name, and all six workers reported success.) - **Whether workers may run checks at all is the lead's call, stated in each brief.** Concurrent scoped test/lint runs are a property of *the project* — a shared database or fixture schema, a fixed port, one build or coverage directory, a lock — so the lead looks, decides, and writes the exact scoped command into each brief or explicitly "none". Either way it runs the full gate once at the end. Commands that *write* count as writing: a repo-wide formatter/autofixer rewrites every worker's files at once, and an emitting build or codegen step has workers racing over shared artifacts — both are the lead's alone, never run while a worker is live. - **Fan-out is not free, and the brief says so.** Every worker pays a full cold start re-reading the repo, and its final message is the only thing that comes back. A sibling project was observed spawning four workers for four one-line edits: four cold starts for work one agent finishes in a single pass. That is the failure the brief argues against, at length, because it is the only lever there is. -- **`CLAUDETM_HIVE_MAX_PARALLEL` (default 10) is a safety ceiling, not a target** (`core/hive.py`). It bounds concurrent workers — one lead plus up to N — and nothing else. A typo in the env var never ends a run: anything unset, unparseable or `<= 0` falls back to the default. +- **`CLAUDETM_HIVE_MAX_PARALLEL` (default 6) is a safety ceiling, not a target** (`core/hive.py`). It bounds concurrent workers — one lead plus up to N — and nothing else. A typo in the env var never ends a run: anything unset, unparseable or `<= 0` falls back to the default. The number is **prompt-visible**: it is interpolated verbatim into the fan-out brief, so changing the constant changes what every lead is told it may dispatch — which is why the default moved from 10 to 6, a width a single shared checkout on one box actually rewards. +- **The ceiling is now enforced, not merely stated.** It had always been prose — a number interpolated into the brief with nothing behind it — and this repo has already measured what a prose-only rule is worth (leads ignored "never background a worker" in 27% of dispatches, which is why that one is pinned on the definition). The bundled CLI enforces its own cap: it hands out concurrency slots and refuses the overflow with `Concurrent subagent limit reached`, reading `CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS` and defaulting to **20** when unset (verified in the bundled binary: `var _e=20; return env.CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS ?? _e`). `_execute_query` now passes `hive_max_parallel()` into the CLI env, so the brief and the runtime finally agree on one number. Related and deliberately *not* set: `CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH` (default 3) would structurally stop a *project specialist* — which claudetm no longer passes definitions for, so it carries no `disallowedTools` — from spawning its own workers, but whether depth 1 also forbids the lead's own dispatch was not verified, and shipping an unverified value that could disable fan-out outright is the worse risk. +- **A fanned-out session shows its team, not just its lines** (`core/hive_roster.py`). Per-worker prefixes answer "who said this"; they cannot answer "how many workers are live, what is each on, and what has each burned" — the questions you actually have while watching a hive. `HiveRoster` tracks each worker from its dispatch block to its returning tool result and renders a compact block: `name#n`, current activity, elapsed, tokens. It is pure state plus a renderer — it prints nothing and imports neither `console` nor the SDK — and `MessageProcessor` is the only thing that turns it into output, forced when the team's shape changes (a dispatch, a return) and otherwise throttled by `CLAUDETM_HIVE_ROSTER_INTERVAL_SEC` (60s; `0` disables). Per-worker tokens come from `AssistantMessage.usage` on messages carrying `parent_tool_use_id` — the only per-subagent figure the stream offers, since the terminal `ResultMessage` aggregates the lead and every worker into the one total that cannot say *which* worker spent it. The SDK does not document that field's exact meaning, so it is best-effort: anything malformed is ignored rather than reported as a number. **A dispatch's tool result is an acknowledgement, not a completion** — measured against a live session, the `ToolResultBlock` for an `Agent` call arrives ~0.1s after the dispatch while that worker's own messages keep arriving for the next 40 seconds. Reading it as "the worker returned" is what produced a roster that jumped straight to `3 done` at `0s` elapsed with all three still working. So only a **failed** dispatch retires a worker (a refused spawn — the CLI's `Concurrent subagent limit reached` — or a crash): that worker never ran, and it is the one completion the block stream states outright. Nothing marks when a *successful* worker stopped, and inventing that signal was the original bug. The roster also deliberately **adopts an unfamiliar tool-use id** (a worker can speak before its spawning block is processed), which makes the call sites load-bearing — handing it every errored top-level `ToolResultBlock` would invent a phantom row per failed Read/Bash call on sessions with no hive at all, so it is gated on an id already recorded as a dispatch. Both are pinned by named regression tests. +- **Within the ceiling, the composition is the lead's and nothing rewards variety.** Up to `max_parallel` workers at once, and they may be N of the same kind, one each of N different kinds, or any mix — four concurrent `hive-worker`s is exactly as legitimate as a `backend-dev`, a `frontend-dev` and two `hive-worker`s. What selects each worker is the piece it is handed: the specialist whose description fits it, else generic `hive-worker`. Two pieces suiting the same specialist get two instances of it. +- **Workers' own prose reaches the log, not just their tool calls** (`CLAUDETM_FORWARD_SUBAGENT_TEXT`, default on). The SDK forwards a subagent's tool calls unasked but withholds its text and thinking behind `forward_subagent_text` (claude-agent-sdk >= 0.2.140), which left a fanned-out session half-visible: you could see that a worker edited a file and never what it was trying to do. The rendering it needs already existed — stable per-worker colour and `#n` ordinal, and subagent text displayed but never accumulated into the lead's result. Forwarding is display-only and costs no tokens; the only price is log volume. - **One checkout, shared by everyone.** Workers work directly in this same tree. Never a git worktree, never a clone, never a per-agent copy — the work has to land in the tree the lead commits from. The exclusive file sets in each worker's brief are the only lock there is; a worker that needs a file it does not own STOPS and reports the collision rather than resolving it silently. - **The lead alone runs git.** Workers read, edit and run narrow checks; they never `add`, `commit`, `branch`, `checkout`, `stash` or `push`. That is what makes the shared checkout safe — one writer to the index, so no `.git/index.lock` contention and no half-staged tree — and it keeps history a single authored series. The lead waits for every worker to return, verifies on disk (`git status`, read the changed files), runs the **full** project gate once, and only then commits. - **A worker's report is not evidence.** The lead never saw its tool calls. A worker that narrated a change without writing it is a real failure mode, so the lead re-checks on disk and re-does or re-assigns that piece itself. diff --git a/README.md b/README.md index c66ba8bb..b445a89b 100644 --- a/README.md +++ b/README.md @@ -383,7 +383,9 @@ Precedence (highest first): real environment variables, then the **active profil | `CLAUDETM_MAX_TURNS` | `400` | Max agent steps per session — a runaway backstop, not a working budget. Set `0` to disable. Overrunning retries the task rather than marking it done | | `CLAUDETM_STREAM_IDLE_TIMEOUT_SEC` | `1800` | Max silence between SDK stream messages before treating the stream as hung | | `CLAUDETM_POST_COMPLETION_IDLE_TIMEOUT_SEC` | `120` | Max wait for the final result message after the agent signals it's done | -| `CLAUDETM_HIVE_MAX_PARALLEL` | `10` | Safety ceiling on concurrent `hive-worker` subagents (1 lead + up to N workers) — a ceiling, not a target; the lead sizes its own team | +| `CLAUDETM_HIVE_MAX_PARALLEL` | `6` | Safety ceiling on concurrent worker subagents (1 lead + up to N workers) — a ceiling, not a target; the lead sizes its own team, and may run N of the same kind, N different kinds, or any mix | +| `CLAUDETM_FORWARD_SUBAGENT_TEXT` | `1` | Stream each worker's own text and thinking into the session log, not just its tool calls. `0` to quiet a fanned-out run | +| `CLAUDETM_HIVE_ROSTER_INTERVAL_SEC` | `60` | How often the live worker roster (who is running, on what, for how long, at what cost) is printed during a fanned-out session. `0` disables it | Sessions are bounded in steps, not wall-clock: a wall-clock cap would punish a slow-but-healthy session (big test suite, slow CI) exactly as hard as a looping one. Use `--budget` for a per-session cost cap. @@ -504,7 +506,7 @@ claudetm config-update --no-parallel # turn it off mid-run **Only the lead runs git.** Workers read, edit and run narrow checks; they never stage, commit, branch or push. The lead waits for every worker to finish, verifies the changes on disk itself, runs the full project gate, and only then commits, pushes and opens the PR — exactly as a single-agent session does. -Cap the fan-out with `CLAUDETM_HIVE_MAX_PARALLEL` (default 10 = one lead plus up to 10 workers). It is a safety ceiling, not a target. +Cap the fan-out with `CLAUDETM_HIVE_MAX_PARALLEL` (default 6 = one lead plus up to 6 workers). It is a safety ceiling, not a target, and within it the composition is the lead's: 6 of the same kind, 6 different kinds, or any mix. Use `--no-parallel` when you want every session strictly single-agent — debugging a run, or a repo where concurrent edits are hard to reason about. diff --git a/pyproject.toml b/pyproject.toml index d88dfc8c..c66385fd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ classifiers = [ ] dependencies = [ - "claude-agent-sdk>=0.2.137,<0.3", + "claude-agent-sdk>=0.2.150,<0.3", "typer>=0.22.0,<1", "pydantic>=2.12.0,<3", "rich>=14.3.2,<16", @@ -41,7 +41,14 @@ Issues = "https://github.com/developerz-ai/claude-task-master/issues" Changelog = "https://github.com/developerz-ai/claude-task-master/blob/main/CHANGELOG.md" [project.optional-dependencies] -mcp = ["mcp>=1.26.0"] +# Upper-bounded on purpose. mcp 2.x removed `mcp.server.fastmcp` (FastMCP was +# renamed to MCPServer), which every module in `claude_task_master/mcp/` imports, +# so a 2.x resolution makes `claudetm-mcp` fail at import. Until claude-agent-sdk +# 0.2.140 the SDK's own `mcp<2.0.0` pin was silently holding this line for us; +# it widened to `mcp<3.0.0`, so a fresh `pip install claude-task-master[mcp]` +# started resolving mcp 2.1.1. Lift this only together with the MCPServer +# migration: https://py.sdk.modelcontextprotocol.io/v2/migration/ +mcp = ["mcp>=1.26.0,<2"] api = [ "fastapi>=0.128.7", "uvicorn[standard]>=0.40.0", diff --git a/requirements.txt b/requirements.txt index 7e7e2036..b3a4f138 100644 --- a/requirements.txt +++ b/requirements.txt @@ -28,7 +28,7 @@ certifi==2026.1.4 # httpx cffi==2.0.0 # via cryptography -claude-agent-sdk==0.1.35 +claude-agent-sdk==0.2.150 # via claude-task-master (pyproject.toml) click==8.3.1 # via @@ -38,6 +38,8 @@ coverage==7.13.4 # via pytest-cov cryptography==46.0.5 # via pyjwt +execnet==2.1.2 + # via pytest-xdist fastapi==0.128.7 # via claude-task-master (pyproject.toml) h11==0.16.0 @@ -63,7 +65,9 @@ idna==3.11 iniconfig==2.3.0 # via pytest jsonschema==4.26.0 - # via mcp + # via + # claude-agent-sdk + # mcp jsonschema-specifications==2025.9.1 # via jsonschema librt==0.7.8 @@ -114,12 +118,15 @@ pytest==9.0.2 # pytest-asyncio # pytest-cov # pytest-timeout + # pytest-xdist pytest-asyncio==1.3.0 # via claude-task-master (pyproject.toml) pytest-cov==7.0.0 # via claude-task-master (pyproject.toml) pytest-timeout==2.4.0 # via claude-task-master (pyproject.toml) +pytest-xdist==3.8.0 + # via claude-task-master (pyproject.toml) python-dotenv==1.2.1 # via # pydantic-settings @@ -140,10 +147,12 @@ rpds-py==0.30.0 # via # jsonschema # referencing -ruff==0.15.0 +ruff==0.16.5 # via claude-task-master (pyproject.toml) shellingham==1.5.4 # via typer +sniffio==1.3.1 + # via claude-agent-sdk sortedcontainers==2.4.0 # via hypothesis sse-starlette==3.2.0 @@ -157,11 +166,15 @@ typer==0.22.0 # via claude-task-master (pyproject.toml) typing-extensions==4.15.0 # via + # anyio # fastapi # mcp # mypy # pydantic # pydantic-core + # pytest-asyncio + # referencing + # starlette # typing-inspection typing-inspection==0.4.2 # via diff --git a/src/claude_task_master/core/agent_error_classify.py b/src/claude_task_master/core/agent_error_classify.py new file mode 100644 index 00000000..2ec0a0b8 --- /dev/null +++ b/src/claude_task_master/core/agent_error_classify.py @@ -0,0 +1,233 @@ +"""Map a raw SDK exception onto a typed :class:`AgentError`. + +The classification decides whether an unattended run **retries or dies**: +:data:`~.agent_exceptions.TRANSIENT_ERRORS` members are retried under the +failure budget, everything else propagates. Getting it wrong in either +direction is expensive, so this module prefers the structured payload the SDK +now hands us and falls back to prose only when there is none. + +Two sources, in order: + +1. **The result payload.** ``claude-agent-sdk`` >= 0.2.140 raises ``ResultError`` + (a ``ProcessError`` subclass) when the CLI exits after a terminal error + result, carrying ``api_error_status``, ``subtype``, ``terminal_reason``, + ``errors`` and ``result``. An HTTP status is a fact; a substring is a guess. + The fields are read with ``getattr`` rather than by importing the SDK — this + module must stay importable with ``claude_agent_sdk`` absent, which the + suite exercises by patching it out of ``sys.modules``. + +2. **The message text**, for every other exception shape. Note that + ``str(ResultError)`` is only ``"Claude Code returned an error result: + (exit code: 1)"`` — the prose that names the actual failure + ("Connection closed mid-response", "API Error: 529 overloaded_error") lives + in ``errors``/``result``. So the payload is folded into the searched text + too, which is what lets a text rule fire on a ``ResultError`` at all. + +The text rules are matched on **token boundaries**, not raw substrings. Two +misclassifications lived in the loose version, both of them one-way: + +- ``"500" in text`` also matched ``"request took 1500ms"``, turning a latency + message into a retryable :class:`APIServerError`. +- ``"auth" in text`` also matched ``"Co-Authored-By"`` — which appears in every + commit message this project writes — turning any error that echoed a git + command into a *non*-retryable :class:`APIAuthenticationError` that ends the + run. +""" + +from __future__ import annotations + +import re +from typing import Any + +from .agent_exceptions import ( + AgentError, + APIAuthenticationError, + APIConnectionError, + APIRateLimitError, + APIServerError, + APITimeoutError, + ContentFilterError, + ModelUnavailableError, + QueryExecutionError, +) + +__all__ = ["classify_api_error", "result_error_fields", "searchable_error_text"] + +#: A standalone HTTP 5xx status, not a digit inside a larger number. +_SERVER_STATUS_RE = re.compile(r"(? dict[str, Any]: + """Structured fields off a ``ResultError``, empty for anything else. + + Duck-typed on purpose: importing ``claude_agent_sdk`` here would make error + classification depend on the SDK being importable, which is exactly the + situation some of these errors describe. + + Args: + error: The raw exception from the SDK stream. + + Returns: + A dict with any of ``api_error_status`` (int), ``subtype``, + ``terminal_reason``, ``result`` (str) and ``errors`` (list[str]) that + the exception actually carries, with wrong-typed values dropped. + """ + fields: dict[str, Any] = {} + + status = getattr(error, "api_error_status", None) + # bool is an int subclass; a True here would read as HTTP status 1. + if isinstance(status, int) and not isinstance(status, bool): + fields["api_error_status"] = status + + for name in ("subtype", "terminal_reason", "result"): + value = getattr(error, name, None) + if isinstance(value, str) and value: + fields[name] = value + + errors = getattr(error, "errors", None) + if isinstance(errors, list): + strings = [e for e in errors if isinstance(e, str) and e] + if strings: + fields["errors"] = strings + + return fields + + +def searchable_error_text(error: Exception, fields: dict[str, Any]) -> str: + """Lower-cased text to match rules against: the message plus its payload. + + ``str(ResultError)`` names only the subtype, so matching on it alone throws + away the one string that says what went wrong. + + Args: + error: The raw exception. + fields: The result of :func:`result_error_fields`. + + Returns: + A single lower-cased string safe to run substring and regex rules over. + """ + parts = [str(error)] + result = fields.get("result") + if result: + parts.append(result) + parts.extend(fields.get("errors", ())) + return "\n".join(parts).lower() + + +def _classify_status(status: int, text: str, error: Exception) -> AgentError | None: + """Map an HTTP status from the result payload onto a typed error. + + Args: + status: ``api_error_status`` as reported by the CLI. + text: The searchable error text, for the one status that needs it. + error: The raw exception, carried through as the typed error's cause so + logs still show what the SDK actually said. + + Returns: + The typed error, or None when the status carries no useful verdict and + the text rules should have their turn. + """ + if status == 429: + return APIRateLimitError(getattr(error, "retry_after", None), error) + if status in (401, 403): + return APIAuthenticationError(error) + if status == 404: + # 404 is Anthropic's answer to an unknown model id. Only a model 404 + # routes to the fallback chain; any other 404 is not a model problem. + if "model" in text: + return ModelUnavailableError(error) + return None + if status in _TIMEOUT_STATUSES: + # A gateway timeout is a timeout first and a 5xx second: the retry is + # the same, but the message should say what actually happened. + return APITimeoutError(_DEFAULT_TIMEOUT_SEC, error) + if 500 <= status <= 599: + # 529 (overloaded) lands here, which is the common real case and one + # the text rules missed entirely — it is absent from str(ResultError). + return APIServerError(status, error) + return None + + +def _classify_text(error: Exception, text: str) -> AgentError: + """Classify from the message text. Always returns something. + + Args: + error: The raw exception, attached to the typed error as its cause. + text: The searchable error text (already lower-cased). + + Returns: + A typed :class:`AgentError`; :class:`QueryExecutionError` when no rule + matches, which the retry layer treats as retryable-but-unclassified. + """ + error_type = type(error).__name__ + + if "content filtering" in text or "output blocked" in text: + return ContentFilterError(error) + + # Require "model" *and* a not-found keyword, so "503 Service Unavailable" + # and "Network unreachable" are not read as a bad model id. + if "model" in text and any( + kw in text + for kw in ("not_found", "not found", "does not exist", "unavailable", "invalid model") + ): + return ModelUnavailableError(error) + + if "rate" in text and "limit" in text: + return APIRateLimitError(getattr(error, "retry_after", None), error) + + if _AUTH_RE.search(text) or _AUTH_STATUS_RE.search(text): + return APIAuthenticationError(error) + + if ( + "timeout" in text + or "timed out" in text + or error_type in ("TimeoutError", "AsyncioTimeoutError") + ): + return APITimeoutError(_DEFAULT_TIMEOUT_SEC, error) + + if any(kw in text for kw in ("connect", "connection", "network")): + return APIConnectionError(error) + + match = _SERVER_STATUS_RE.search(text) + if match: + return APIServerError(int(match.group(1)), error) + + return QueryExecutionError(f"API error: {error}", error) + + +def classify_api_error(error: Exception) -> AgentError: + """Classify a raw SDK exception into a typed :class:`AgentError`. + + Args: + error: The exception raised out of the SDK query stream. + + Returns: + A typed error. Membership in ``TRANSIENT_ERRORS`` is what decides + whether the caller retries. + """ + fields = result_error_fields(error) + text = searchable_error_text(error, fields) + + status = fields.get("api_error_status") + if isinstance(status, int): + classified = _classify_status(status, text, error) + if classified is not None: + return classified + + return _classify_text(error, text) diff --git a/src/claude_task_master/core/agent_message.py b/src/claude_task_master/core/agent_message.py index 029d049b..a0a16d51 100644 --- a/src/claude_task_master/core/agent_message.py +++ b/src/claude_task_master/core/agent_message.py @@ -23,7 +23,9 @@ from typing import TYPE_CHECKING, Any from . import console +from .agent_message_roster import _MessageRosterMixin from .console import SubagentPalette +from .hive_roster import HiveRoster if TYPE_CHECKING: from .logger import TaskLogger @@ -58,7 +60,7 @@ def _subagent_tool_use_id(message: Any) -> str | None: return None -class MessageProcessor: +class MessageProcessor(_MessageRosterMixin): """Handles processing of messages from the Claude Agent SDK query stream. This class is responsible for parsing messages, displaying tool usage, @@ -95,6 +97,11 @@ def __init__(self, logger: "TaskLogger | None" = None): # the module-level ``console``) so tests that patch ``console`` to # capture output still exercise the real labelling. self._subagent_palette = SubagentPalette() + # Team-level view of a fanned-out session. The per-worker prefixes show + # each line as it happens; this answers the question they cannot — how + # many workers are live right now, what each is on, and what each has + # burned. Pure state plus a renderer: it prints nothing itself. + self._roster = HiveRoster() def reset_result_state(self) -> None: """Clear captured terminal-result state before a new query. @@ -110,6 +117,7 @@ def reset_result_state(self) -> None: self.last_output_tokens = 0 self._subagent_names = {} self._subagent_palette.clear() + self._roster.clear() def _note_subagent_spawn(self, block: Any, tool_input: Any) -> None: """Record the subagent name behind a Task/Agent tool call. @@ -135,7 +143,28 @@ def _note_subagent_spawn(self, block: Any, tool_input: Any) -> None: # line, so ``#n`` counts workers in the order the lead spawned them # — the order the reader just watched go past — instead of in the # order they happened to speak. - self._subagent_palette.slot(tool_use_id) + ordinal = self._subagent_palette.slot(tool_use_id) + self._roster_note_dispatch(tool_use_id, name, ordinal) + + def _render_roster(self, *, force: bool = False) -> None: + """Print the team view, if one is due and there is a team to show. + + Lives here rather than on the roster mixin so that it resolves the same + ``console`` the rest of this processor prints through — a second import + in another module is a second thing to patch, and the streamed-output + tests rightly patch exactly one. + + Args: + force: Bypass the interval (a worker started, or a dispatch failed). + """ + if not self._roster.due(force=force): + return + lines = self._roster.render() + if not lines: + return + console.newline() + for line in lines: + console.detail(line) def _stream_prefix(self, subagent_id: str | None) -> str: """Console-only marker distinguishing subagent output from our own. @@ -193,6 +222,8 @@ def process_message(self, message: Any, result_text: str) -> str: if block_type == "TextBlock": # Claude's text response - show with [claude] prefix console.claude_text(f"{prefix}{block.text.strip()}", flush=True) + if subagent_id is not None: + self._roster.note_activity(subagent_id, block.text) # A subagent's narration is visible but is not this # session's output: accumulating it lets a worker's prose # (a stray "TASK COMPLETE", say) speak for the lead. @@ -205,10 +236,20 @@ def process_message(self, message: Any, result_text: str) -> str: tool_detail = self.format_tool_detail(block.name, tool_input) console.tool(f"{prefix}Using tool: {block.name} {tool_detail}", flush=True) self._note_subagent_spawn(block, tool_input) + if subagent_id is not None: + self._roster.note_activity( + subagent_id, f"{block.name} {tool_detail}".strip() + ) # Log to file if logger is available if self.logger: self.logger.log_tool_use(block.name, tool_input) elif block_type == "ToolResultBlock": + # A top-level result for a dispatch we are tracking is that + # worker returning. Only the lead's own stream can close a + # worker out: the id inside a subagent's stream is one of + # *its* tool calls, not the worker itself. + if subagent_id is None: + self._roster_note_dispatch_result(block) # Tool result - show completion with [claude] prefix if block.is_error: console.tool_result(f"{prefix}Tool error", is_error=True) @@ -219,6 +260,9 @@ def process_message(self, message: Any, result_text: str) -> str: if self.logger: self.logger.log_tool_result(block.tool_use_id, "completed") + if subagent_id is not None: + self._roster_note_worker_message(subagent_id, message) + # Handle RateLimitEvent typed messages from SDK v0.1.49+ if message_type == "RateLimitEvent": retry_after = getattr(message, "retry_after", None) diff --git a/src/claude_task_master/core/agent_message_roster.py b/src/claude_task_master/core/agent_message_roster.py new file mode 100644 index 00000000..37c1d6b6 --- /dev/null +++ b/src/claude_task_master/core/agent_message_roster.py @@ -0,0 +1,98 @@ +"""Roster wiring for :class:`~.agent_message.MessageProcessor`. + +Separate from the message processor for two reasons. It is a second reason to +change that file (how a fanned-out team is *summarised*, as against how a single +message is rendered), and the rules below are the load-bearing part of the +feature — they are what the live SDK stream actually says, as opposed to what it +looks like it should say. Keeping them here keeps both files inside the 500-LOC +house limit with room for the explanation. + +:class:`HiveRoster` itself is deliberately tolerant: it adopts an unfamiliar +tool-use id on sight, because a worker can speak before the block that spawned +it has been processed. That tolerance moves the burden here — feeding it the +wrong ids is what invents workers that never existed. +""" + +from __future__ import annotations + +from typing import Any + +from .hive_roster import HiveRoster + +__all__ = ["_MessageRosterMixin"] + + +class _MessageRosterMixin: + """Feeds and renders the hive roster from the SDK message stream. + + Attribute stubs satisfy mypy; concrete values come from MessageProcessor. + """ + + _roster: HiveRoster + _subagent_names: dict[str, str] + + def _render_roster(self, *, force: bool = False) -> None: + """Print the roster — implemented by MessageProcessor.""" + raise NotImplementedError # pragma: no cover + + def _roster_note_dispatch(self, tool_use_id: str, name: str, ordinal: int) -> None: + """Put a freshly dispatched worker on the roster and show it at once. + + Args: + tool_use_id: The spawning tool-use id, this worker's identity. + name: The ``subagent_type`` dispatched. + ordinal: The palette slot, so ``#n`` matches the streamed prefixes. + """ + self._roster.spawn(tool_use_id, name, ordinal) + self._render_roster(force=True) + + def _roster_note_dispatch_result(self, block: Any) -> None: + """Close out a worker whose *dispatch itself* failed. Nothing else. + + A dispatch's tool result is an **acknowledgement, not a completion**. + Measured against a live session: the ``ToolResultBlock`` for an ``Agent`` + call arrives 0.1s after the dispatch, while that worker's own messages + keep arriving for the next 40 seconds. Reading it as "the worker + returned" marked every worker done on arrival — a roster that jumped + straight to ``3 done`` at ``0s`` elapsed while all three were still + working. + + So only a *failed* dispatch finishes a worker: an error here means the + spawn was refused (the CLI's "Concurrent subagent limit reached") or + crashed, so that worker never ran — the one completion this stream + states outright. A successful worker simply stays live. Nothing in the + block stream marks when it stopped, and inventing that signal is what + produced the wrong roster to begin with. + + Args: + block: A top-level ``ToolResultBlock``. + """ + if not getattr(block, "is_error", False): + return + tool_use_id = getattr(block, "tool_use_id", None) + # A known dispatch id only: every ordinary Read/Bash/Grep result reaches + # here too, and the roster would adopt each one as a worker. + if not isinstance(tool_use_id, str) or tool_use_id not in self._subagent_names: + return + self._roster.finish(tool_use_id, is_error=True) + self._render_roster(force=True) + + def _roster_note_worker_message(self, subagent_id: str, message: Any) -> None: + """Attribute a worker's tokens to it, then refresh the view if due. + + ``AssistantMessage.usage`` on a message carrying ``parent_tool_use_id`` + is the only per-subagent figure the stream offers: the terminal + ``ResultMessage`` aggregates the lead and every worker into a single + total, which is precisely the number that cannot answer *which* worker + spent it. The SDK does not document the per-message field's meaning, so + this is best-effort by construction — the roster ignores anything + malformed rather than reporting a figure it cannot stand up. + + Args: + subagent_id: The message's ``parent_tool_use_id``. + message: The message from that worker. + """ + usage = getattr(message, "usage", None) + if usage is not None: + self._roster.add_usage(subagent_id, usage) + self._render_roster() diff --git a/src/claude_task_master/core/agent_query_execute.py b/src/claude_task_master/core/agent_query_execute.py index f197aeb5..125621ed 100644 --- a/src/claude_task_master/core/agent_query_execute.py +++ b/src/claude_task_master/core/agent_query_execute.py @@ -22,6 +22,7 @@ StreamStallError, WorkingDirectoryError, ) +from .hive import forward_subagent_text_enabled, hive_max_parallel if TYPE_CHECKING: from .agent_models import ModelType @@ -149,6 +150,19 @@ async def _execute_query( cli_env = { "CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": stall_timeout_ms, "API_TIMEOUT_MS": stall_timeout_ms, + # Make the hive ceiling structural instead of merely stated. The + # number has always been interpolated into the fan-out brief, but + # prose was the only thing holding it: nothing stopped a lead from + # dispatching twelve workers, and this repo has already measured + # what a prose-only rule is worth (leads ignored the + # "never background a worker" instruction in 27% of dispatches, + # which is why that one is now pinned on the definition). + # + # The CLI enforces this one itself — it hands out concurrency slots + # and refuses the overflow dispatch with "Concurrent subagent limit + # reached", defaulting to 20 when unset. Handing it our own ceiling + # means the brief and the runtime finally agree on one number. + "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": str(hive_max_parallel()), } # Inject the active profile's auth context (isolated CLAUDE_CONFIG_DIR # for oauth profiles, or ANTHROPIC_API_KEY/BASE_URL for api-key @@ -189,6 +203,18 @@ async def _execute_query( # passes no loader gets no dispatch. if get_agents_func is None: options_kwargs["disallowed_tools"] = list(DISPATCH_TOOLS) + else: + # This session may fan out, so its workers' own prose is worth + # having. The SDK forwards a subagent's tool calls unasked but + # not its text or thinking, which left the hive half-visible: + # you could see that a worker edited a file and never what it + # was trying to do. The renderer for it already exists — each + # worker's lines carry a stable colour and ``#n`` ordinal, and + # subagent text is deliberately not accumulated into the lead's + # own result (see ``MessageProcessor.process_message``), so the + # only thing this changes is what reaches the log. + if forward_subagent_text_enabled(): + options_kwargs["forward_subagent_text"] = True # Add effort level for extended thinking depth control if effort_level: diff --git a/src/claude_task_master/core/agent_query_helpers.py b/src/claude_task_master/core/agent_query_helpers.py index 4a405344..17e31c65 100644 --- a/src/claude_task_master/core/agent_query_helpers.py +++ b/src/claude_task_master/core/agent_query_helpers.py @@ -4,24 +4,14 @@ - :meth:`_default_get_model_name` — maps ModelType to API model name string - :meth:`_default_process_message` — accumulates text from SDK stream messages -- :meth:`_classify_api_error` — maps raw exceptions to typed AgentError subclasses +- :meth:`_classify_api_error` — thin delegator to :mod:`.agent_error_classify` """ from __future__ import annotations from typing import TYPE_CHECKING, Any -from .agent_exceptions import ( - AgentError, - APIAuthenticationError, - APIConnectionError, - APIRateLimitError, - APIServerError, - APITimeoutError, - ContentFilterError, - ModelUnavailableError, - QueryExecutionError, -) +from .agent_exceptions import AgentError from .config_loader import get_config if TYPE_CHECKING: @@ -92,60 +82,21 @@ def _default_process_message(self, message: Any, result_text: str) -> str: def _classify_api_error(self, error: Exception) -> AgentError: """Classify an API error into a specific error type. + Delegates to :func:`~.agent_error_classify.classify_api_error`. The + policy lives in its own module because the verdict decides whether an + unattended run retries or dies, and because it now has two sources to + reconcile — the SDK's structured ``ResultError`` payload and the message + text — which is more than one reason for this mixin to change. + Args: error: The original exception. Returns: A classified AgentError subclass. """ - error_str = str(error).lower() - error_type = type(error).__name__ - - # Check for content filtering errors (not retryable) - if "content filtering" in error_str or "output blocked" in error_str: - return ContentFilterError(error) - - # Check for model-availability errors (recover via fallback chain, not by - # retrying the same model). Anthropic returns not_found_error for an - # unknown/unavailable model id. Require both "model" and a not-found - # keyword so generic messages like "503 Service Unavailable" or - # "Network unreachable" are not misclassified. - if "model" in error_str and any( - kw in error_str - for kw in ("not_found", "not found", "does not exist", "unavailable", "invalid model") - ): - return ModelUnavailableError(error) - - # Check for rate limiting - if "rate" in error_str and "limit" in error_str: - # Try to extract retry-after if present - retry_after = None - if hasattr(error, "retry_after"): - retry_after = error.retry_after - return APIRateLimitError(retry_after, error) - - # Check for authentication errors - if any(kw in error_str for kw in ["auth", "unauthorized", "403", "401"]): - return APIAuthenticationError(error) - - # Check for timeout errors - if "timeout" in error_str or error_type in ("TimeoutError", "AsyncioTimeoutError"): - return APITimeoutError(30.0, error) - - # Check for connection errors - if any(kw in error_str for kw in ["connect", "connection", "network"]): - return APIConnectionError(error) - - # Check for server errors (5xx) - if "500" in error_str or "502" in error_str or "503" in error_str or "504" in error_str: - # Try to extract status code - for code in [500, 502, 503, 504]: - if str(code) in error_str: - return APIServerError(code, error) - return APIServerError(500, error) - - # Default to generic query execution error - return QueryExecutionError(f"API error: {error}", error) + from .agent_error_classify import classify_api_error # noqa: PLC0415 + + return classify_api_error(error) __all__ = ["_AgentQueryHelpersMixin"] diff --git a/src/claude_task_master/core/hive.py b/src/claude_task_master/core/hive.py index 3ff56fd7..acc65b68 100644 --- a/src/claude_task_master/core/hive.py +++ b/src/claude_task_master/core/hive.py @@ -25,14 +25,17 @@ from typing import Literal, cast __all__ = [ + "DEFAULT_FORWARD_SUBAGENT_TEXT", "DEFAULT_HIVE_MAX_PARALLEL", "DEFAULT_HIVE_WORKER_EFFORT", "DEFAULT_HIVE_WORKER_MAX_TURNS", + "FORWARD_SUBAGENT_TEXT_ENV", "HIVE_MAX_PARALLEL_ENV", "HIVE_WORKER_EFFORT_ENV", "HIVE_WORKER_MAX_TURNS_ENV", "describe_machine", "fan_out_enabled", + "forward_subagent_text_enabled", "hive_max_parallel", "hive_worker_effort", "hive_worker_max_turns", @@ -46,11 +49,35 @@ _VALID_EFFORTS: frozenset[str] = frozenset({"low", "medium", "high", "xhigh", "max"}) -# One knob: 1 lead + up to this many workers running at once. -DEFAULT_HIVE_MAX_PARALLEL: int = 10 +#: One knob: 1 lead + up to this many workers running at once. +#: +#: Six, not ten. The ceiling is interpolated verbatim into the fan-out brief +#: (``prompts_working_hive``), so this number is what the lead is told it may +#: dispatch — it is a prompt-visible constant, not merely an internal guard. +#: Ten invited a split wider than a shared checkout rewards: every worker is a +#: full agent process re-reading this repo on the same cores, and past a handful +#: the cold starts and the lead's own verification pass cost more than the +#: concurrency returns. It stays a *ceiling*, never a target — the brief says so +#: at length, and zero remains the right answer for most tasks. +DEFAULT_HIVE_MAX_PARALLEL: int = 6 HIVE_MAX_PARALLEL_ENV = "CLAUDETM_HIVE_MAX_PARALLEL" +#: Whether a fanned-out session streams its workers' own text and thinking. +#: +#: The SDK forwards a subagent's *tool calls* to the parent stream on its own, +#: but not its prose — so a lead's ``↳ [hive-worker#2]`` lines showed what a +#: worker touched and never what it was doing or why. ``forward_subagent_text`` +#: (claude-agent-sdk >= 0.2.140) closes that gap, and the rendering it needs +#: already existed here: per-worker colour, a stable ``#n`` ordinal, and the +#: rule that a worker's text is displayed but never accumulated into the lead's +#: own result. Forwarding is display-only — it costs no tokens, because those +#: blocks were generated either way; the only price is log volume, which is the +#: thing a fanned-out session was most missing. +DEFAULT_FORWARD_SUBAGENT_TEXT: bool = True + +FORWARD_SUBAGENT_TEXT_ENV = "CLAUDETM_FORWARD_SUBAGENT_TEXT" + #: Per-worker turn budget, passed as ``AgentDefinition.maxTurns``. #: #: A worker is meant to take a whole module or feature slice off the lead's @@ -124,6 +151,40 @@ def _env_positive_int(name: str, default: int) -> int: return value if value > 0 else default +def _env_bool(name: str, default: bool) -> bool: + """Read a boolean from the environment, falling back on anything unclear. + + Never raises. A typo in an env var must not change a run's semantics, so + only the recognised spellings count and everything else defers to *default* + rather than being read as a silent "off". + + Args: + name: Environment variable to read. + default: Value to use when unset or unrecognised. + + Returns: + The parsed flag, or *default*. + """ + raw = os.environ.get(name) + if raw is None: + return default + value = raw.strip().lower() + if value in ("1", "true", "yes", "on"): + return True + if value in ("0", "false", "no", "off"): + return False + return default + + +def forward_subagent_text_enabled() -> bool: + """Whether to stream workers' own text and thinking into the session log. + + Reads ``CLAUDETM_FORWARD_SUBAGENT_TEXT``; anything unset or unrecognised + falls back to :data:`DEFAULT_FORWARD_SUBAGENT_TEXT`. + """ + return _env_bool(FORWARD_SUBAGENT_TEXT_ENV, DEFAULT_FORWARD_SUBAGENT_TEXT) + + def hive_max_parallel() -> int: """Maximum concurrent hive workers (1 lead + up to N workers). diff --git a/src/claude_task_master/core/hive_roster.py b/src/claude_task_master/core/hive_roster.py new file mode 100644 index 00000000..a42f7a72 --- /dev/null +++ b/src/claude_task_master/core/hive_roster.py @@ -0,0 +1,499 @@ +"""Live roster of the workers a lead session has running — the hive, at a glance. + +A fanned-out work session interleaves several ``hive-worker`` subagents into one +stream. :class:`~.console.SubagentPalette` already makes an individual line +attributable (``↳ [hive-worker#2] …``), which answers "who said this?" — but a +human watching the run still has no view of the *team*: how many workers are +alive, what each is doing right now, how long each has been at it, and what each +has burned. On an unattended run that is the difference between "the session has +been quiet for four minutes" and "four workers are mid-edit and one finished". + +This module owns that view and nothing else: it is a state tracker plus a +renderer, and it **returns lines rather than printing them**. Printing here +would put a second ANSI/colour policy next to ``console``'s, the one place that +decides whether escapes are emitted at all (a redirected run writes +``.claude-task-master/logs/``, where escapes are noise) — and a renderer that +prints can only be tested by capturing stdout, while one returning ``list[str]`` +is tested by reading it. The caller in ``agent_message`` does the printing. + +The output is deliberately **line-oriented, not a TUI**. Claude Code repaints a +roster in place with cursor control; claudetm's console is an append-only stream +that is routinely piped into a file, so a repainting roster would render there +as a screenful of garbage per update. Instead the block is emitted whole, at +most once per :data:`DEFAULT_ROSTER_INTERVAL_SEC` (see :meth:`HiveRoster.due`) — +a periodic snapshot reads correctly both live and in a log a day later. + +Nothing here may raise. It renders diagnostics for a run nobody is watching; a +crash in the status display taking down a session that was working fine would be +strictly worse than showing no status at all. Every entry point tolerates +unknown ids, missing fields and garbage values. +""" + +from __future__ import annotations + +import math +import os +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +__all__ = [ + "DEFAULT_ROSTER_INTERVAL_SEC", + "MAX_ACTIVITY_CHARS", + "ROSTER_INTERVAL_ENV", + "HiveRoster", + "format_elapsed", + "format_tokens", + "roster_interval_sec", +] + + +#: How often the roster block may be emitted, in seconds. +#: +#: A fan-out runs for many minutes, so this is a *heartbeat*, not an animation: +#: often enough that a stalled team is visible within a minute, rare enough that +#: the block never competes with the workers' own output for the reader's +#: attention. ``0`` in the env var turns the roster off entirely. +DEFAULT_ROSTER_INTERVAL_SEC: int = 60 + +ROSTER_INTERVAL_ENV = "CLAUDETM_HIVE_ROSTER_INTERVAL_SEC" + +#: Width the "current activity" cell is truncated to. +#: +#: An activity string is whatever the worker last did or said, which can be a +#: paragraph. The roster is a scan-in-one-glance table; a cell that wraps makes +#: every column below it meaningless, so the text is cut rather than allowed to +#: reflow the block. +MAX_ACTIVITY_CHARS: int = 60 + +#: Name given to a worker adopted from an update seen before its dispatch — a +#: real subagent whose ``subagent_type`` we have not been told yet. +_UNKNOWN_NAME = "subagent" + + +def roster_interval_sec() -> int: + """Seconds between roster emissions; ``0`` means the roster is disabled. + + Reads :data:`ROSTER_INTERVAL_ENV` on **every call**, never at import, as the + repo already does elsewhere (``_env_positive_int`` in :mod:`.hive`): a + malformed env var must degrade to the default rather than raise at import + and take down every claudetm command, ``status`` and ``doctor`` included. + Unlike ``hive``'s helper this one cannot reject ``0``, because ``0`` is a + meaningful setting here — "never show the roster", for anyone who finds the + block noisy. Negative and unparseable values still fall back to the default. + + Returns: + The interval in seconds, or ``0`` when the roster is switched off. + """ + raw = os.environ.get(ROSTER_INTERVAL_ENV) + if raw is None: + return DEFAULT_ROSTER_INTERVAL_SEC + try: + value = int(raw.strip()) + except (ValueError, AttributeError, TypeError): + return DEFAULT_ROSTER_INTERVAL_SEC + if value == 0: + return 0 + return value if value > 0 else DEFAULT_ROSTER_INTERVAL_SEC + + +def format_elapsed(seconds: float) -> str: + """Render a duration compactly: ``47s``, ``19m 50s``, ``1h 04m``. + + One unit pair, largest first, seconds dropped past the hour — the reader is + asking "is this worker stuck?", and at that scale the seconds digit is + noise. Minor components are zero-padded (``1m 02s``) so a column of these + stays aligned without the caller padding it. + + The clock is :func:`time.monotonic` in production, but a negative or + non-finite value can still arrive (timestamps from two sources, a stepped + clock in a test). Both render as ``0s`` rather than ``-1m 60s`` or ``nans``: + a wrong-looking zero is readable, and raising here would kill a session over + a cosmetic. + + Args: + seconds: Elapsed wall time. + + Returns: + A short human-readable duration, never empty. + """ + try: + value = float(seconds) + except (TypeError, ValueError): + return "0s" + if not math.isfinite(value) or value < 0: + return "0s" + total = int(value) + if total < 60: + return f"{total}s" + if total < 3600: + return f"{total // 60}m {total % 60:02d}s" + return f"{total // 3600}h {(total % 3600) // 60:02d}m" + + +def format_tokens(count: int) -> str: + """Render a token count compactly: ``847``, ``410.2k``, ``1.2M``. + + A worker's cold start alone runs to hundreds of thousands of tokens, so raw + digits make the column both wide and hard to compare at a glance. One decimal + place is the useful resolution: the number answers "is this worker + expensive?", not "exactly how expensive". + + Args: + count: A token count. Negative, non-integral or unusable values render + as ``"0"`` rather than raising. + + Returns: + A short human-readable count, never empty. + """ + try: + value = int(count) + except (TypeError, ValueError, OverflowError): + return "0" + if value < 0: + return "0" + if value < 1000: + return str(value) + # 999_950 is where one decimal place would round up to "1000.0k"; hand it to + # the next unit instead so the column never gains a digit. + if value < 999_950: + return f"{value / 1000:.1f}k" + return f"{value / 1_000_000:.1f}M" + + +def _normalize_activity(activity: str) -> str: + """Collapse an activity string to one short single-line cell. + + Worker text arrives as prose with newlines and runs of spaces; a row is one + line, so whitespace is collapsed *before* truncation (the other order spends + the budget on indentation). Blank input returns ``""``, which the caller + reads as "no update" and keeps the previous activity — a worker that emits + an empty block has not stopped doing what it was doing. + """ + try: + collapsed = " ".join(str(activity).split()) + except (TypeError, ValueError): + return "" + if not collapsed: + return "" + if len(collapsed) <= MAX_ACTIVITY_CHARS: + return collapsed + return collapsed[: MAX_ACTIVITY_CHARS - 1].rstrip() + "…" + + +def _usage_int(usage: Any, key: str) -> int: + """Pull one integer token field out of an SDK usage payload. + + ``usage`` is a plain dict in the Python SDK, but other shapes carry + attributes — :mod:`.agent_message` learned this the expensive way (a + ``getattr`` on a dict returned 0, so every cost report read zero tokens), so + both are accepted here from the start. Anything missing, ``None`` or + unparseable contributes 0: an incomplete payload degrades the figure, never + the run. + """ + if usage is None: + return 0 + raw: Any + if isinstance(usage, dict): + raw = usage.get(key) + else: + raw = getattr(usage, key, None) + if raw is None or isinstance(raw, bool): + return 0 + try: + value = int(raw) + except (TypeError, ValueError, OverflowError): + return 0 + return value if value > 0 else 0 + + +@dataclass +class _Worker: + """One dispatched subagent's live state. + + Private: the roster's public surface is its methods and its rendered lines, + so nothing outside can come to depend on this shape. + """ + + tool_use_id: str + name: str + ordinal: int + started: float + activity: str = "" + input_tokens: int = 0 + output_tokens: int = 0 + ended: float | None = None + is_error: bool = False + + @property + def label(self) -> str: + """``"#"`` with a 1-based ordinal, matching the stream marker. + + The roster and the per-line ``↳ [hive-worker#2]`` marker must name a + worker identically or the reader cannot connect the two. + """ + return f"{self.name}#{self.ordinal + 1}" + + +class HiveRoster: + """Tracks the workers a lead has dispatched and renders them as a block. + + Keyed by ``tool_use_id``, exactly as :class:`~.console.SubagentPalette` is + and for the same reason: the concurrent workers a lead spawns share one + agent name, so the name identifies a *role*, not an instance. The id is + minted at dispatch and constant for that worker's whole life, which is what + lets a row keep its identity across updates. An instance is owned by one + message processor and :meth:`clear`ed between queries, so ordinals restart + per session rather than growing across a run. + + Finished workers stay in the block until ``clear()``. They cost one line and + answer what the live rows cannot — "did the ones that already returned + succeed, and what did they cost?" — which is exactly what a reader needs + when a fan-out ends with a dirty tree. + """ + + def __init__(self, *, clock: Callable[[], float] | None = None) -> None: + """Initialize an empty roster. + + Args: + clock: Monotonic time source, injected so tests can step time + without sleeping (this repo's pytest runs under ``--timeout=2``, + so a test that actually waits is a flake waiting to happen). + Defaults to :func:`time.monotonic`, which is immune to the wall + clock being adjusted mid-session. + """ + self._clock: Callable[[], float] = clock or time.monotonic + self._workers: dict[str, _Worker] = {} + self._last_emit: float | None = None + + def clear(self) -> None: + """Forget every worker and reset the emission timer (between queries).""" + self._workers = {} + self._last_emit = None + + def spawn(self, tool_use_id: str, name: str, ordinal: int) -> None: + """Record that the lead dispatched a worker. + + **Idempotent.** The same dispatch can be seen more than once (a replayed + tool block, a caller that called :meth:`note_activity` first), and + re-registering would restart the worker's clock and zero its tokens — + turning a worker eight minutes into its piece into one that looks fresh, + the single most misleading thing this display could do. + + Args: + tool_use_id: The dispatching tool-use id; the row's identity. + name: The ``subagent_type`` — ``"hive-worker"``, or a project + specialist such as ``"backend-dev"``. + ordinal: 0-based slot from ``SubagentPalette``, displayed 1-based. + """ + existing = self._workers.get(tool_use_id) + if existing is not None: + # Only ever *improve* what we know; never reset the clock. + if name and existing.name == _UNKNOWN_NAME: + existing.name = name + return + self._workers[tool_use_id] = _Worker( + tool_use_id=tool_use_id, + name=name or _UNKNOWN_NAME, + ordinal=ordinal if ordinal >= 0 else self._next_ordinal(), + started=self._now(), + ) + + def note_activity(self, tool_use_id: str, activity: str) -> None: + """Replace what this worker is shown to be doing right now. + + Called on every tool use and every text block from the worker, so the + cell always shows the most recent thing — a roster that accumulated + history would be a log, and the stream is already the log. + + Args: + tool_use_id: The worker's dispatching tool-use id. Unknown ids are + adopted rather than dropped: a worker can speak before its + dispatching tool block has been processed, and a row named + ``subagent`` is far better than a silently missing worker. + activity: Free text; whitespace-collapsed and truncated. Blank + input is ignored so the previous activity survives. + """ + text = _normalize_activity(activity) + worker = self._ensure(tool_use_id) + if text: + worker.activity = text + + def add_usage(self, tool_use_id: str, usage: Any) -> None: + """Accumulate this worker's token usage. + + Cache tokens are summed **into the input total**. A worker's cold start + is almost entirely cache reads — it re-reads the repo it was handed — so + a figure counting only ``input_tokens`` reports a worker that burned half + a million tokens as having burned a few thousand, useless as the one + number that could justify or condemn a fan-out. + + Args: + tool_use_id: The worker's dispatching tool-use id (adopted if new). + usage: An SDK usage payload — dict or attribute-carrying object. + ``None``, missing fields and garbage values all add 0. + """ + worker = self._ensure(tool_use_id) + worker.input_tokens += ( + _usage_int(usage, "input_tokens") + + _usage_int(usage, "cache_read_input_tokens") + + _usage_int(usage, "cache_creation_input_tokens") + ) + worker.output_tokens += _usage_int(usage, "output_tokens") + + def finish(self, tool_use_id: str, *, is_error: bool = False) -> None: + """Mark a worker as returned; its elapsed time freezes here. + + The row stays in :meth:`render` until :meth:`clear` — see the class + docstring for why a finished worker is still worth a line. + + Args: + tool_use_id: The worker's dispatching tool-use id (adopted if new). + is_error: True when the worker's tool result was an error, which + changes only the row's mark and label. The lead still has to + verify every worker's claim on disk; this is a hint to a human, + never a verdict. + """ + worker = self._ensure(tool_use_id) + if worker.ended is None: + worker.ended = self._now() + worker.is_error = worker.is_error or is_error + + @property + def live_count(self) -> int: + """How many workers are dispatched and have not yet returned.""" + return sum(1 for w in self._workers.values() if w.ended is None) + + def due(self, *, force: bool = False) -> bool: + """Whether the caller should emit the roster now (rate limiter). + + The roster exists to be glanceable, so it must not print on every stream + message; the caller asks this and prints only when told to. + + Two deliberate asymmetries. A disabled roster (interval ``0``) answers + False even to ``force`` — "off" a caller can override is not off. And an + *empty* roster answers False without consuming the interval, so a caller + polling before any worker exists cannot spend the first window on a + block that would have rendered as nothing. + + Args: + force: Emit regardless of how recently the last block went out — + for the moments worth showing on sight, such as the first + dispatch or the last worker returning. + + Returns: + True at most once per interval (see :func:`roster_interval_sec`). + """ + interval = roster_interval_sec() + if interval <= 0 or not self._workers: + return False + now = self._now() + if force or self._last_emit is None or (now - self._last_emit) >= interval: + self._last_emit = now + return True + return False + + def render(self) -> list[str]: + """Render the roster block as plain lines — no ANSI, no emoji. + + Colour is the caller's business (:mod:`.console` owns that policy) and + emoji are banned from this project's printer outright, so status is + carried by ASCII marks: ``~`` running, ``+`` done, ``!`` failed. + + Column widths are measured from the rows rather than fixed, because a + project specialist's name is arbitrarily long — one + ``database-migration-specialist#3`` would push every following column + out of alignment in a hardcoded layout. + + Returns: + The block's lines (header first, one line per worker, ordinal + order), or ``[]`` when there is nothing to show or the roster is + disabled — callers can print the result unconditionally. + """ + if roster_interval_sec() <= 0 or not self._workers: + return [] + now = self._now() + rows = [self._row(w, now) for w in sorted(self._workers.values(), key=_sort_key)] + widths = [max(len(row[i]) for row in rows) for i in range(1, 5)] + lines = [self._header()] + for mark, label, activity, elapsed, tok_in, tok_out in rows: + lines.append( + f" {mark} {label:<{widths[0]}} {activity:<{widths[1]}} " + f"{elapsed:>{widths[2]}} in {tok_in:>{widths[3]}} out {tok_out}".rstrip() + ) + return lines + + # --- internals ---------------------------------------------------- + + def _now(self) -> float: + """Read the injected clock, tolerating one that misbehaves.""" + try: + return float(self._clock()) + except Exception: # A broken clock must never end a run. + return 0.0 + + def _next_ordinal(self) -> int: + """The ordinal to give a worker we were never told the slot of.""" + if not self._workers: + return 0 + return max(w.ordinal for w in self._workers.values()) + 1 + + def _ensure(self, tool_use_id: str) -> _Worker: + """Fetch a worker's row, adopting an unknown id as a new one. + + Adoption rather than rejection: the alternative is dropping the update, + and a demonstrably alive worker (it just spoke) missing from the roster + is the one failure this display cannot afford. + """ + worker = self._workers.get(tool_use_id) + if worker is not None: + return worker + worker = _Worker( + tool_use_id=tool_use_id, + name=_UNKNOWN_NAME, + ordinal=self._next_ordinal(), + started=self._now(), + ) + self._workers[tool_use_id] = worker + return worker + + def _header(self) -> str: + """The count line: how many are working, how many have returned.""" + live = self.live_count + done = len(self._workers) - live + parts: list[str] = [] + if live: + parts.append(f"{live} {_plural(live, 'worker')} active") + if done: + parts.append(f"{done} done") + return "Hive: " + ", ".join(parts) + + def _row(self, worker: _Worker, now: float) -> tuple[str, str, str, str, str, str]: + """One worker's cells: mark, label, activity, elapsed, in, out.""" + if worker.ended is None: + mark = "~" + activity = worker.activity or "starting" + elapsed = now - worker.started + else: + mark = "!" if worker.is_error else "+" + activity = "failed" if worker.is_error else "done" + elapsed = worker.ended - worker.started + return ( + mark, + worker.label, + activity, + format_elapsed(elapsed), + format_tokens(worker.input_tokens), + format_tokens(worker.output_tokens), + ) + + +def _sort_key(worker: _Worker) -> tuple[int, str]: + """Order rows by dispatch slot, so a worker never moves between blocks.""" + return (worker.ordinal, worker.tool_use_id) + + +def _plural(count: int, noun: str) -> str: + """``"worker"`` / ``"workers"`` — English only; this is a console line.""" + return noun if count == 1 else f"{noun}s" diff --git a/src/claude_task_master/core/prompts_working_hive.py b/src/claude_task_master/core/prompts_working_hive.py index 88726b97..722c98a5 100644 --- a/src/claude_task_master/core/prompts_working_hive.py +++ b/src/claude_task_master/core/prompts_working_hive.py @@ -106,7 +106,14 @@ def build_fanout_section( **{max_parallel} concurrent workers is a safety ceiling, not a target.** Pick the number of genuinely disjoint pieces your task actually has — often zero — and never pad a split to look -parallel.{machine_note}{_project_agents_note(project_agents)} +parallel. + +**Within that ceiling the composition is entirely yours.** Up to {max_parallel} workers running at +once, and they may be {max_parallel} of the same kind, one each of {max_parallel} different kinds, +or any mix. Four concurrent `hive-worker`s is exactly as legitimate as one `backend-dev`, one +`frontend-dev` and two `hive-worker`s — nothing rewards variety for its own sake. What selects each +worker is the piece it is being handed: the specialist whose description fits that piece, else a +generic `hive-worker`. Two pieces that suit the same specialist get two instances of it.{machine_note}{_project_agents_note(project_agents)} **Dispatch is part of the speed** — the Agent tool: a project specialist where one fits, else `subagent_type: "hive-worker"`, at most {max_parallel} workers at a time. Put every independent diff --git a/tests/core/test_agent_error_classify.py b/tests/core/test_agent_error_classify.py new file mode 100644 index 00000000..64e50145 --- /dev/null +++ b/tests/core/test_agent_error_classify.py @@ -0,0 +1,523 @@ +"""Tests for agent_error_classify - mapping raw SDK exceptions onto AgentError. + +This module contains tests for: +- The structured path (``ResultError``'s ``api_error_status`` and friends) +- ``result_error_fields`` / ``searchable_error_text`` field extraction +- The text path (token-boundary rules over the message plus its payload) +- The two loose-substring regressions the module exists to prevent +- Importability with ``claude_agent_sdk`` absent + +The classification decides whether an unattended run retries or dies, so each +case asserts the concrete type - membership in TRANSIENT_ERRORS is what the +retry layer branches on. +""" + +import importlib +import inspect +from typing import Any +from unittest.mock import patch + +import pytest + +from claude_task_master.core import agent_error_classify +from claude_task_master.core.agent_error_classify import ( + classify_api_error, + result_error_fields, + searchable_error_text, +) +from claude_task_master.core.agent_exceptions import ( + TRANSIENT_ERRORS, + APIAuthenticationError, + APIConnectionError, + APIRateLimitError, + APIServerError, + APITimeoutError, + ContentFilterError, + ModelUnavailableError, + QueryExecutionError, +) + +# ============================================================================= +# Helpers +# ============================================================================= + + +class FakeResultError(Exception): + """Duck-typed stand-in for the SDK's ResultError. + + The module reads the payload with getattr rather than by importing the SDK, + so the main path is exercised with a fake. One test below pins the same + contract against a genuinely constructed ``claude_agent_sdk.ResultError``. + """ + + def __init__(self, message: str = "", **payload: Any): + super().__init__(message) + for name, value in payload.items(): + setattr(self, name, value) + + +#: What str(ResultError) actually looks like - it names the subtype and nothing +#: else, which is why the payload has to be folded into the searched text. +TERMINAL_MESSAGE = "Claude Code returned an error result: error_during_execution (exit code: 1)" + + +# ============================================================================= +# result_error_fields Tests +# ============================================================================= + + +class TestResultErrorFields: + """Tests for result_error_fields payload extraction.""" + + def test_plain_exception_yields_empty_dict(self): + """Test a plain Exception carries no structured payload.""" + assert result_error_fields(Exception("boom")) == {} + + def test_all_fields_extracted(self): + """Test every supported field is picked up off the exception.""" + error = FakeResultError( + TERMINAL_MESSAGE, + api_error_status=529, + subtype="error_during_execution", + terminal_reason="api_error", + result="API Error: 529 overloaded_error", + errors=["Connection closed mid-response"], + ) + assert result_error_fields(error) == { + "api_error_status": 529, + "subtype": "error_during_execution", + "terminal_reason": "api_error", + "result": "API Error: 529 overloaded_error", + "errors": ["Connection closed mid-response"], + } + + def test_status_as_string_is_dropped(self): + """Test a wrong-typed api_error_status is ignored, not coerced.""" + error = FakeResultError("boom", api_error_status="429") + assert "api_error_status" not in result_error_fields(error) + + def test_bool_status_is_dropped(self): + """Test a bool api_error_status is not read as HTTP status 1. + + bool is an int subclass, so a naive isinstance(status, int) would let + True through and classify it as a status the mapper has no verdict for + - or worse, a real one. + """ + error = FakeResultError("boom", api_error_status=True) + assert "api_error_status" not in result_error_fields(error) + + def test_errors_not_a_list_is_dropped(self): + """Test a non-list errors field is ignored.""" + error = FakeResultError("boom", errors="Connection closed mid-response") + assert "errors" not in result_error_fields(error) + + def test_non_string_entries_inside_errors_are_dropped(self): + """Test only string entries survive inside errors.""" + error = FakeResultError("boom", errors=[1, None, "real one", ""]) + assert result_error_fields(error)["errors"] == ["real one"] + + def test_all_entries_dropped_leaves_no_errors_key(self): + """Test an errors list with nothing usable in it is omitted entirely.""" + error = FakeResultError("boom", errors=[None, 0, ""]) + assert "errors" not in result_error_fields(error) + + def test_empty_strings_are_dropped(self): + """Test empty subtype/result/terminal_reason values are omitted.""" + error = FakeResultError("boom", subtype="", terminal_reason="", result="") + assert result_error_fields(error) == {} + + +# ============================================================================= +# searchable_error_text Tests +# ============================================================================= + + +class TestSearchableErrorText: + """Tests for searchable_error_text.""" + + def test_plain_exception_is_just_its_message(self): + """Test text for a payload-less error is the lower-cased message.""" + error = Exception("Some UNKNOWN Error") + assert searchable_error_text(error, {}) == "some unknown error" + + def test_payload_prose_is_folded_in(self): + """Test result and errors prose joins the searched text. + + str(ResultError) names only the subtype; the string that says what + actually failed lives in result/errors. Folding it in is what lets a + text rule fire on a ResultError at all. + """ + error = FakeResultError( + TERMINAL_MESSAGE, + result="API Error: 529 overloaded_error", + errors=["Connection closed mid-response"], + ) + text = searchable_error_text(error, result_error_fields(error)) + assert "529 overloaded_error" in text + assert "connection closed mid-response" in text + + def test_text_is_lower_cased(self): + """Test payload prose is lower-cased along with the message.""" + error = FakeResultError("BOOM", errors=["Connection CLOSED"]) + text = searchable_error_text(error, result_error_fields(error)) + assert text == "boom\nconnection closed" + + def test_subtype_and_reason_are_not_searched_as_prose(self): + """Test only result/errors are folded in, per the module contract.""" + error = FakeResultError("boom", subtype="error_during_execution") + text = searchable_error_text(error, result_error_fields(error)) + assert text == "boom" + + +# ============================================================================= +# Structured Path (api_error_status) Tests +# ============================================================================= + + +class TestStructuredStatusClassification: + """Tests for classification driven by the result payload's HTTP status.""" + + @pytest.mark.parametrize( + ("status", "expected", "status_code"), + [ + (429, APIRateLimitError, None), + (401, APIAuthenticationError, None), + (403, APIAuthenticationError, None), + (408, APITimeoutError, None), + (504, APITimeoutError, None), + (500, APIServerError, 500), + (503, APIServerError, 503), + (529, APIServerError, 529), + ], + ) + def test_status_maps_to_type(self, status, expected, status_code): + """Test each decisive status maps to its typed error. + + 529 is Anthropic's "overloaded" status and the real-world case the old + text rules missed entirely: it never appears in str(ResultError), so + only the structured payload can see it. + """ + error = FakeResultError(TERMINAL_MESSAGE, api_error_status=status) + classified = classify_api_error(error) + assert isinstance(classified, expected) + if status_code is not None: + assert classified.status_code == status_code + + def test_504_is_a_timeout_not_a_server_error(self): + """Test a gateway timeout is reported as a timeout, not a bare 5xx.""" + error = FakeResultError(TERMINAL_MESSAGE, api_error_status=504) + classified = classify_api_error(error) + assert isinstance(classified, APITimeoutError) + assert not isinstance(classified, APIServerError) + + def test_529_is_transient(self): + """Test an overloaded API is retryable, which is the whole point.""" + error = FakeResultError(TERMINAL_MESSAGE, api_error_status=529) + assert isinstance(classify_api_error(error), TRANSIENT_ERRORS) + + def test_404_with_model_in_payload_is_model_unavailable(self): + """Test a model 404 routes to the fallback chain.""" + error = FakeResultError( + TERMINAL_MESSAGE, + api_error_status=404, + result="API Error: 404 model claude-nope not_found", + ) + assert isinstance(classify_api_error(error), ModelUnavailableError) + + def test_404_without_model_falls_through_to_text_rules(self): + """Test a non-model 404 is not treated as a bad model id.""" + error = FakeResultError( + TERMINAL_MESSAGE, + api_error_status=404, + result="API Error: 404 no such endpoint", + ) + classified = classify_api_error(error) + assert not isinstance(classified, ModelUnavailableError) + assert isinstance(classified, QueryExecutionError) + + @pytest.mark.parametrize("status", [400, 418]) + def test_status_without_a_verdict_falls_through_to_text_rules(self, status): + """Test a status the mapper has no opinion on defers to the prose.""" + error = FakeResultError( + TERMINAL_MESSAGE, + api_error_status=status, + errors=["Connection closed mid-response"], + ) + assert isinstance(classify_api_error(error), APIConnectionError) + + def test_bool_status_is_ignored_by_the_classifier(self): + """Test api_error_status=True does not classify as HTTP status 1.""" + error = FakeResultError("Some unknown error", api_error_status=True) + classified = classify_api_error(error) + assert type(classified) is QueryExecutionError + + def test_connection_prose_in_errors_classifies_as_connection(self): + """Test payload-only prose reaches the text rules. + + "Connection closed mid-response" lives in ``errors`` and nowhere in + str(error); it must still classify as a retryable connection error. + """ + error = FakeResultError( + TERMINAL_MESSAGE, + subtype="error_during_execution", + errors=["Connection closed mid-response"], + ) + assert isinstance(classify_api_error(error), APIConnectionError) + + @pytest.mark.parametrize("status", [429, 401, 403, 408, 504, 500, 529]) + def test_structured_path_keeps_the_raw_exception(self, status): + """Test the raw exception stays reachable on the structured path.""" + error = FakeResultError(TERMINAL_MESSAGE, api_error_status=status) + classified = classify_api_error(error) + assert isinstance(classified, QueryExecutionError) + assert classified.original_error is error + + def test_text_path_keeps_the_raw_exception(self): + """Test the raw exception stays reachable on the text path too.""" + error = Exception("Connection refused") + classified = classify_api_error(error) + assert isinstance(classified, QueryExecutionError) + assert classified.original_error is error + + +# ============================================================================= +# Real claude_agent_sdk.ResultError Tests +# ============================================================================= + + +class TestRealResultError: + """Pins the contract against a genuinely constructed SDK ResultError.""" + + @staticmethod + def _build() -> Any: + """Build the real ResultError the SDK raises for an API failure.""" + result_error = pytest.importorskip("claude_agent_sdk").ResultError + return result_error( + "Claude Code returned an error result: error_during_execution", + data={ + "subtype": "error_during_execution", + "errors": ["Connection closed mid-response"], + "result": "API Error: 529 overloaded_error", + "api_error_status": 529, + "terminal_reason": "api_error", + }, + exit_code=1, + ) + + def test_str_names_only_the_subtype(self): + """Test the assumption the module is built on: str() says nothing useful.""" + error = self._build() + assert str(error) == TERMINAL_MESSAGE + assert "529" not in str(error) + + def test_fields_are_extracted_from_the_real_class(self): + """Test result_error_fields reads the SDK's own attributes.""" + fields = result_error_fields(self._build()) + assert fields["api_error_status"] == 529 + assert fields["subtype"] == "error_during_execution" + assert fields["terminal_reason"] == "api_error" + assert fields["result"] == "API Error: 529 overloaded_error" + assert fields["errors"] == ["Connection closed mid-response"] + + def test_classifies_as_a_retryable_server_error(self): + """Test a real overloaded-API ResultError is retried, not fatal.""" + error = self._build() + classified = classify_api_error(error) + assert isinstance(classified, APIServerError) + assert classified.status_code == 529 + assert isinstance(classified, TRANSIENT_ERRORS) + assert classified.original_error is error + + def test_real_result_error_without_status_uses_payload_prose(self): + """Test a real ResultError with no status classifies off errors prose.""" + result_error = pytest.importorskip("claude_agent_sdk").ResultError + error = result_error( + "Claude Code returned an error result: error_during_execution", + data={ + "subtype": "error_during_execution", + "errors": ["Connection closed mid-response"], + }, + exit_code=1, + ) + assert isinstance(classify_api_error(error), APIConnectionError) + + +# ============================================================================= +# Text Path Tests +# ============================================================================= + + +class TestTextClassification: + """Tests for classification from the message text alone.""" + + @pytest.mark.parametrize( + ("message", "expected"), + [ + ("API rate limit exceeded", APIRateLimitError), + ("Rate limit: too many requests", APIRateLimitError), + ("HTTP 401 Unauthorized", APIAuthenticationError), + ("HTTP 403 Forbidden", APIAuthenticationError), + ("Unauthorized access to API", APIAuthenticationError), + ("Invalid API key provided", APIAuthenticationError), + ("Request timeout after 30 seconds", APITimeoutError), + ("Request timed out", APITimeoutError), + ("Connection refused", APIConnectionError), + ("Network unreachable", APIConnectionError), + ("Connection failed to server", APIConnectionError), + ("Output blocked by content filtering policy", ContentFilterError), + ("model claude-nope not_found", ModelUnavailableError), + ("The model does not exist", ModelUnavailableError), + ], + ) + def test_message_maps_to_type(self, message, expected): + """Test today's text rules keep classifying as they always did.""" + assert isinstance(classify_api_error(Exception(message)), expected) + + @pytest.mark.parametrize("status", [500, 502, 503]) + def test_server_status_in_text(self, status): + """Test a standalone 5xx in the message yields the right status_code.""" + classified = classify_api_error(Exception(f"HTTP {status} Something Failed")) + assert isinstance(classified, APIServerError) + assert classified.status_code == status + + def test_case_insensitive(self): + """Test the rules ignore case, as the old classifier did.""" + for message in ("rate limit exceeded", "RATE LIMIT EXCEEDED", "Rate Limit Exceeded"): + assert isinstance(classify_api_error(Exception(message)), APIRateLimitError) + + def test_unmatched_message_is_plain_query_execution_error(self): + """Test an unrecognised message stays unclassified rather than guessed.""" + error = Exception("Some unknown error") + classified = classify_api_error(error) + assert type(classified) is QueryExecutionError + assert classified.original_error is error + + def test_empty_message_is_plain_query_execution_error(self): + """Test an empty message matches nothing.""" + assert type(classify_api_error(Exception(""))) is QueryExecutionError + + def test_timeout_error_type_classifies_by_its_type_name(self): + """Test a TimeoutError is a timeout even with an unhelpful message.""" + classified = classify_api_error(TimeoutError("boom")) + assert isinstance(classified, APITimeoutError) + + +# ============================================================================= +# Rule Ordering Tests +# ============================================================================= + + +class TestRuleOrdering: + """Tests that one rule's keywords do not get stolen by another's.""" + + def test_service_unavailable_is_not_a_model_problem(self): + """Test "503 Service Unavailable" is a server error, not a bad model id. + + The model rule requires "model" AND a not-found keyword; matching + "unavailable" alone would send a transient 503 down the fallback chain. + """ + classified = classify_api_error(Exception("HTTP 503 Service Unavailable")) + assert not isinstance(classified, ModelUnavailableError) + assert isinstance(classified, APIServerError) + assert classified.status_code == 503 + + def test_network_unreachable_is_not_a_model_problem(self): + """Test "Network unreachable" is a connection error, not a bad model id.""" + classified = classify_api_error(Exception("Network unreachable")) + assert not isinstance(classified, ModelUnavailableError) + assert isinstance(classified, APIConnectionError) + + def test_content_filter_wins_over_a_status_in_the_same_message(self): + """Test a filtered response is not read as a plain HTTP failure.""" + message = "API Error: 400 content filtering blocked the response" + assert isinstance(classify_api_error(Exception(message)), ContentFilterError) + + +# ============================================================================= +# Regression Tests +# ============================================================================= + + +class TestLooseSubstringRegressions: + """The two misclassifications this module exists to prevent.""" + + def test_latency_figure_is_not_a_server_error(self): + """Regression: `"500" in text` matched "request took 1500ms". + + A latency figure inside a larger number turned an ordinary message into + a retryable APIServerError. The status rule is anchored on digit + boundaries now, so 1500 is not 500. + """ + classified = classify_api_error(Exception("request took 1500ms")) + assert not isinstance(classified, APIServerError) + assert type(classified) is QueryExecutionError + + def test_larger_numbers_containing_a_5xx_are_not_server_errors(self): + """Regression, generalised: only a standalone 5xx counts as a status.""" + for message in ("elapsed 1500ms", "wrote 5031 bytes", "offset 25036"): + assert not isinstance(classify_api_error(Exception(message)), APIServerError) + + def test_co_authored_by_trailer_is_not_an_auth_failure(self): + """Regression: `"auth" in text` matched "Co-Authored-By". + + Every commit message this project writes carries that trailer, so any + error echoing a git command became a NON-retryable + APIAuthenticationError and ended the unattended run. Auth phrasings are + whole words now. + """ + message = ( + "Command failed: git commit -m 'fix: thing\n\n" + "Co-Authored-By: Claude '" + ) + classified = classify_api_error(Exception(message)) + assert not isinstance(classified, APIAuthenticationError) + + def test_co_authored_by_in_result_payload_is_not_an_auth_failure(self): + """Regression: the same trailer arriving via the structured payload.""" + error = FakeResultError( + TERMINAL_MESSAGE, + errors=["git commit failed\nCo-Authored-By: Claude "], + ) + assert not isinstance(classify_api_error(error), APIAuthenticationError) + + def test_author_and_authorize_words_are_not_auth_failures(self): + """Regression, generalised: "auth" inside a longer word never matches.""" + for message in ("Author: sebi", "authoring the changelog entry"): + assert not isinstance(classify_api_error(Exception(message)), APIAuthenticationError) + + +# ============================================================================= +# Importability Tests +# ============================================================================= + + +class TestImportableWithoutSDK: + """The module must classify errors with claude_agent_sdk absent. + + Some of the errors it classifies describe exactly that situation, so it + duck-types the payload instead of importing the SDK. + """ + + def test_classification_works_with_sdk_patched_out(self): + """Test import and classification with claude_agent_sdk unimportable.""" + with patch.dict("sys.modules", {"claude_agent_sdk": None}): + reloaded = importlib.reload(agent_error_classify) + try: + error = FakeResultError(TERMINAL_MESSAGE, api_error_status=529) + classified = reloaded.classify_api_error(error) + assert isinstance(classified, APIServerError) + assert classified.status_code == 529 + assert isinstance( + reloaded.classify_api_error(Exception("Connection refused")), + APIConnectionError, + ) + finally: + importlib.reload(agent_error_classify) + + def test_module_does_not_import_the_sdk(self): + """Test the SDK is not imported - the payload is read with getattr.""" + assert "claude_agent_sdk" not in vars(agent_error_classify) + source = inspect.getsource(agent_error_classify) + code = "\n".join(line for line in source.splitlines() if not line.lstrip().startswith("#")) + assert "import claude_agent_sdk" not in code + assert "from claude_agent_sdk" not in code diff --git a/tests/core/test_agent_message.py b/tests/core/test_agent_message.py index 104d45aa..395da0da 100644 --- a/tests/core/test_agent_message.py +++ b/tests/core/test_agent_message.py @@ -1124,3 +1124,119 @@ def test_dict_usage_missing_keys_keep_zeros(self): assert proc.last_input_tokens == 0 assert proc.last_output_tokens == 0 + + +# ============================================================================= +# Hive roster wiring — the team view of a fanned-out session +# ============================================================================= + + +class TestHiveRosterWiring: + """The roster is fed from the stream, and only from the right parts of it. + + The roster itself is deliberately tolerant — it adopts an unfamiliar + tool-use id on sight, because a worker can speak before the block that + spawned it has been processed. That tolerance makes the *call sites* the + thing under test: feeding it the wrong ids is what invents workers that + never existed. + """ + + def test_dispatch_registers_a_worker(self): + """A Task/Agent tool call puts that worker on the roster.""" + proc = MessageProcessor() + block = _make_tool_use_block("Agent", {"subagent_type": "hive-worker"}, block_id="toolu_w1") + proc.process_message(_make_assistant_message([block]), "") + + assert proc._roster.live_count == 1 + assert any("hive-worker#1" in line for line in proc._roster.render()) + + def test_ordinary_tool_results_do_not_invent_workers(self): + """Regression: a top-level Read/Bash result is not a worker returning. + + ``finish()`` adopts an unknown id, so passing it every top-level + ToolResultBlock produced a phantom ``subagent#n done`` row per tool + call — on a solo session with no hive at all. Only an id we recorded as + a dispatch may close a worker out. + """ + proc = MessageProcessor() + for tool_id in ("toolu_read_1", "toolu_bash_2", "toolu_grep_3"): + proc.process_message(_make_assistant_message([_make_tool_result_block(tool_id)]), "") + # A *failing* ordinary tool call is the dangerous one: only errored + # results reach finish(), so that is where a phantom row would appear. + proc.process_message( + _make_assistant_message([_make_tool_result_block("toolu_bash_4", is_error=True)]), + "", + ) + + assert proc._roster.live_count == 0 + assert proc._roster.render() == [] + + def test_successful_dispatch_result_does_not_finish_the_worker(self): + """Regression: a dispatch's tool result is an ACK, not a completion. + + Measured against a live session: the ``ToolResultBlock`` for an + ``Agent`` call arrives ~0.1s after the dispatch, while that worker's + own messages keep arriving for the next 40 seconds. Treating it as the + worker returning produced a roster that jumped straight to ``3 done`` + at ``0s`` elapsed with all three still working. + """ + proc = MessageProcessor() + spawn = _make_tool_use_block("Agent", {"subagent_type": "hive-worker"}, block_id="toolu_w1") + proc.process_message(_make_assistant_message([spawn]), "") + assert proc._roster.live_count == 1 + + proc.process_message(_make_assistant_message([_make_tool_result_block("toolu_w1")]), "") + assert proc._roster.live_count == 1, "an ack must not retire a live worker" + + def test_failed_dispatch_finishes_the_worker(self): + """A refused spawn is the one completion the stream states outright. + + The CLI refuses an over-ceiling dispatch with "Concurrent subagent + limit reached"; that worker never runs, so it must not sit on the + roster as live for the rest of the session. + """ + proc = MessageProcessor() + spawn = _make_tool_use_block("Agent", {"subagent_type": "hive-worker"}, block_id="toolu_w1") + proc.process_message(_make_assistant_message([spawn]), "") + + proc.process_message( + _make_assistant_message([_make_tool_result_block("toolu_w1", is_error=True)]), "" + ) + assert proc._roster.live_count == 0 + assert any("failed" in line for line in proc._roster.render()) + + def test_subagent_activity_and_usage_are_attributed(self): + """A worker's own lines update its activity and token totals.""" + proc = MessageProcessor() + spawn = _make_tool_use_block("Agent", {"subagent_type": "hive-worker"}, block_id="toolu_w1") + proc.process_message(_make_assistant_message([spawn]), "") + + msg = _make_subagent_message( + [_make_text_block("rewriting the migration tests")], + parent_tool_use_id="toolu_w1", + ) + msg.usage = {"input_tokens": 4000, "output_tokens": 250} + proc.process_message(msg, "") + + rendered = "\n".join(proc._roster.render()) + assert "rewriting the migration tests" in rendered + assert "4.0k" in rendered + + def test_subagent_text_is_not_accumulated_into_the_lead_result(self): + """Roster wiring must not disturb the existing accumulation rule.""" + proc = MessageProcessor() + msg = _make_subagent_message( + [_make_text_block("TASK COMPLETE")], parent_tool_use_id="toolu_w1" + ) + assert proc.process_message(msg, "lead text") == "lead text" + + def test_reset_clears_the_roster(self): + """A reused processor must not carry a prior session's team forward.""" + proc = MessageProcessor() + spawn = _make_tool_use_block("Agent", {"subagent_type": "hive-worker"}, block_id="toolu_w1") + proc.process_message(_make_assistant_message([spawn]), "") + assert proc._roster.live_count == 1 + + proc.reset_result_state() + assert proc._roster.live_count == 0 + assert proc._roster.render() == [] diff --git a/tests/core/test_hive.py b/tests/core/test_hive.py index 6823f99e..90748c5c 100644 --- a/tests/core/test_hive.py +++ b/tests/core/test_hive.py @@ -17,9 +17,12 @@ import pytest from claude_task_master.core.hive import ( + DEFAULT_FORWARD_SUBAGENT_TEXT, DEFAULT_HIVE_MAX_PARALLEL, + FORWARD_SUBAGENT_TEXT_ENV, HIVE_MAX_PARALLEL_ENV, describe_machine, + forward_subagent_text_enabled, hive_max_parallel, ) @@ -71,7 +74,13 @@ def test_env_read_at_call_time(self, monkeypatch: pytest.MonkeyPatch) -> None: assert hive_max_parallel() == 7 def test_default_value(self) -> None: - assert DEFAULT_HIVE_MAX_PARALLEL == 10 + """Six, and pinned: the number is interpolated into the fan-out brief. + + ``DEFAULT_HIVE_MAX_PARALLEL`` is not an internal guard — it is rendered + verbatim into the lead's prompt as the ceiling it may dispatch up to, so + changing it silently changes what every lead is told. + """ + assert DEFAULT_HIVE_MAX_PARALLEL == 6 # ============================================================================= @@ -135,3 +144,47 @@ def test_a_broken_probe_never_raises(self, monkeypatch: pytest.MonkeyPatch) -> N raising=False, ) assert isinstance(describe_machine("/nonexistent-path-for-test"), str) + + +# ============================================================================= +# forward_subagent_text_enabled - workers' own prose in the session log +# ============================================================================= + + +class TestForwardSubagentTextEnabled: + """Failure cases first: an unrecognised value must never flip the flag. + + Reading a typo as "off" would silently blind a fanned-out run, which is the + exact opposite of what the knob is for — so anything unclear keeps the + default rather than being coerced to False. + """ + + def test_unset_uses_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(FORWARD_SUBAGENT_TEXT_ENV, raising=False) + assert forward_subagent_text_enabled() is DEFAULT_FORWARD_SUBAGENT_TEXT + + @pytest.mark.parametrize("raw", ["garbage", "", " ", "2", "maybe", "None"]) + def test_unrecognised_falls_back(self, monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv(FORWARD_SUBAGENT_TEXT_ENV, raw) + assert forward_subagent_text_enabled() is DEFAULT_FORWARD_SUBAGENT_TEXT + + @pytest.mark.parametrize("raw", ["0", "false", "FALSE", "no", "off", " Off "]) + def test_off_spellings(self, monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv(FORWARD_SUBAGENT_TEXT_ENV, raw) + assert forward_subagent_text_enabled() is False + + @pytest.mark.parametrize("raw", ["1", "true", "TRUE", "yes", "on", " On "]) + def test_on_spellings(self, monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv(FORWARD_SUBAGENT_TEXT_ENV, raw) + assert forward_subagent_text_enabled() is True + + def test_default_is_on(self) -> None: + """A fanned-out session is worth watching; opting out is the exception.""" + assert DEFAULT_FORWARD_SUBAGENT_TEXT is True + + def test_env_read_at_call_time(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Not frozen at import, like every other knob in this module.""" + monkeypatch.setenv(FORWARD_SUBAGENT_TEXT_ENV, "off") + assert forward_subagent_text_enabled() is False + monkeypatch.setenv(FORWARD_SUBAGENT_TEXT_ENV, "on") + assert forward_subagent_text_enabled() is True diff --git a/tests/core/test_hive_roster.py b/tests/core/test_hive_roster.py new file mode 100644 index 00000000..2d768d1a --- /dev/null +++ b/tests/core/test_hive_roster.py @@ -0,0 +1,640 @@ +"""Tests for the hive roster — the live view of a lead's worker team. + +Failure cases first, because this module renders diagnostics for an unattended +run: a crash in the status display would take down a session that was otherwise +working, which is strictly worse than showing no status. So every entry point is +tested with an unknown id, a missing field and outright garbage before anything +is tested with well-formed input. + +Time is injected, never slept on. The suite runs under ``--timeout=2`` with +several agents on the box, so a test that waits on a real clock is a flake; the +fake clock also makes "a worker has been running for four minutes" a one-line +setup instead of an impossibility. +""" + +from __future__ import annotations + +import math + +import pytest + +from claude_task_master.core.hive_roster import ( + DEFAULT_ROSTER_INTERVAL_SEC, + MAX_ACTIVITY_CHARS, + ROSTER_INTERVAL_ENV, + HiveRoster, + format_elapsed, + format_tokens, + roster_interval_sec, +) + + +class FakeClock: + """A monotonic clock the test drives by hand.""" + + def __init__(self, start: float = 1000.0) -> None: + self.now = start + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture(autouse=True) +def _default_interval(monkeypatch: pytest.MonkeyPatch) -> None: + """Run every test against the default interval unless it says otherwise. + + The env var is process-wide, so a developer with it set would otherwise see + a different suite than CI does. + """ + monkeypatch.delenv(ROSTER_INTERVAL_ENV, raising=False) + + +@pytest.fixture +def clock() -> FakeClock: + return FakeClock() + + +@pytest.fixture +def roster(clock: FakeClock) -> HiveRoster: + return HiveRoster(clock=clock) + + +# ============================================================================= +# roster_interval_sec - a typo must never end an unattended run +# ============================================================================= + + +class TestRosterIntervalSec: + def test_default_when_unset(self) -> None: + assert roster_interval_sec() == DEFAULT_ROSTER_INTERVAL_SEC + + def test_default_value(self) -> None: + assert DEFAULT_ROSTER_INTERVAL_SEC == 60 + + def test_reads_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "15") + assert roster_interval_sec() == 15 + + def test_whitespace_tolerated(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, " 30 ") + assert roster_interval_sec() == 30 + + def test_zero_disables(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Unlike hive's knob, 0 is meaningful here: turn the roster off.""" + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "0") + assert roster_interval_sec() == 0 + + def test_negative_falls_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "-5") + assert roster_interval_sec() == DEFAULT_ROSTER_INTERVAL_SEC + + def test_garbage_falls_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "often") + assert roster_interval_sec() == DEFAULT_ROSTER_INTERVAL_SEC + + def test_empty_falls_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "") + assert roster_interval_sec() == DEFAULT_ROSTER_INTERVAL_SEC + + def test_float_falls_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "2.5") + assert roster_interval_sec() == DEFAULT_ROSTER_INTERVAL_SEC + + def test_env_read_at_call_time(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Not frozen at import: a test (or a run) may set it per-process.""" + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "10") + assert roster_interval_sec() == 10 + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "20") + assert roster_interval_sec() == 20 + + +# ============================================================================= +# format_elapsed - never negative, never absurd +# ============================================================================= + + +class TestFormatElapsed: + def test_negative_is_zero(self) -> None: + """A clock that went backwards renders 0s, not a negative duration.""" + assert format_elapsed(-42.0) == "0s" + + def test_nan_is_zero(self) -> None: + assert format_elapsed(math.nan) == "0s" + + def test_infinity_is_zero(self) -> None: + assert format_elapsed(math.inf) == "0s" + + def test_garbage_is_zero(self) -> None: + assert format_elapsed("later") == "0s" # type: ignore[arg-type] + + def test_zero(self) -> None: + assert format_elapsed(0) == "0s" + + def test_seconds(self) -> None: + assert format_elapsed(47.9) == "47s" + + def test_just_under_a_minute(self) -> None: + assert format_elapsed(59.999) == "59s" + + def test_exactly_a_minute(self) -> None: + assert format_elapsed(60) == "1m 00s" + + def test_minutes_pad_seconds(self) -> None: + assert format_elapsed(62) == "1m 02s" + + def test_minutes(self) -> None: + assert format_elapsed(19 * 60 + 50) == "19m 50s" + + def test_just_under_an_hour(self) -> None: + assert format_elapsed(3599) == "59m 59s" + + def test_exactly_an_hour(self) -> None: + assert format_elapsed(3600) == "1h 00m" + + def test_hours_drop_seconds(self) -> None: + assert format_elapsed(3600 + 4 * 60 + 59) == "1h 04m" + + def test_many_hours(self) -> None: + assert format_elapsed(26 * 3600) == "26h 00m" + + +# ============================================================================= +# format_tokens - a worker's cold start is six figures +# ============================================================================= + + +class TestFormatTokens: + def test_negative_is_zero(self) -> None: + assert format_tokens(-1) == "0" + + def test_garbage_is_zero(self) -> None: + assert format_tokens("many") == "0" # type: ignore[arg-type] + + def test_none_is_zero(self) -> None: + assert format_tokens(None) == "0" # type: ignore[arg-type] + + def test_zero(self) -> None: + assert format_tokens(0) == "0" + + def test_small(self) -> None: + assert format_tokens(847) == "847" + + def test_just_under_a_thousand(self) -> None: + assert format_tokens(999) == "999" + + def test_exactly_a_thousand(self) -> None: + assert format_tokens(1000) == "1.0k" + + def test_thousands(self) -> None: + assert format_tokens(410_249) == "410.2k" + + def test_rounding_never_yields_four_digit_k(self) -> None: + """999_999 must not render as '1000.0k' and widen the column.""" + assert format_tokens(999_999) == "1.0M" + + def test_millions(self) -> None: + assert format_tokens(1_240_000) == "1.2M" + + +# ============================================================================= +# spawn / finish lifecycle +# ============================================================================= + + +class TestLifecycle: + def test_empty_roster_has_no_live_workers(self, roster: HiveRoster) -> None: + assert roster.live_count == 0 + + def test_spawn_counts_live(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "hive-worker", 1) + assert roster.live_count == 2 + + def test_finish_drops_out_of_live_count(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "hive-worker", 1) + roster.finish("t1") + assert roster.live_count == 1 + + def test_finished_worker_still_rendered(self, roster: HiveRoster) -> None: + """A returned worker's cost and outcome is exactly what a reader wants.""" + roster.spawn("t1", "hive-worker", 0) + roster.finish("t1") + assert any("hive-worker#1" in line for line in roster.render()) + + def test_finish_freezes_elapsed(self, roster: HiveRoster, clock: FakeClock) -> None: + roster.spawn("t1", "hive-worker", 0) + clock.advance(62) + roster.finish("t1") + clock.advance(9999) + assert any("1m 02s" in line for line in roster.render()) + + def test_finish_is_idempotent(self, roster: HiveRoster, clock: FakeClock) -> None: + roster.spawn("t1", "hive-worker", 0) + clock.advance(62) + roster.finish("t1") + clock.advance(300) + roster.finish("t1") + assert any("1m 02s" in line for line in roster.render()) + + def test_finish_unknown_id_does_not_raise(self, roster: HiveRoster) -> None: + roster.finish("never-seen") + assert roster.live_count == 0 + assert any("subagent#1" in line for line in roster.render()) + + def test_error_finish_marked(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.finish("t1", is_error=True) + row = next(line for line in roster.render() if "hive-worker#1" in line) + assert row.lstrip().startswith("!") + assert "failed" in row + + def test_clear_forgets_everything(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.clear() + assert roster.live_count == 0 + assert roster.render() == [] + + def test_default_clock_is_usable(self) -> None: + """No injected clock: still works, without the test depending on time.""" + default = HiveRoster() + default.spawn("t1", "hive-worker", 0) + assert default.live_count == 1 + assert default.render() + + +class TestSpawnIdempotence: + def test_respawn_does_not_restart_the_clock(self, roster: HiveRoster, clock: FakeClock) -> None: + """A worker running for four minutes must not read as freshly started.""" + roster.spawn("t1", "hive-worker", 0) + clock.advance(240) + roster.spawn("t1", "hive-worker", 0) + assert any("4m 00s" in line for line in roster.render()) + + def test_respawn_does_not_zero_tokens(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": 5000, "output_tokens": 2000}) + roster.spawn("t1", "hive-worker", 0) + row = next(line for line in roster.render() if "hive-worker#1" in line) + assert "in 5.0k out 2.0k" in row + + def test_respawn_does_not_duplicate_the_row(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t1", "hive-worker", 0) + assert roster.live_count == 1 + assert len(roster.render()) == 2 # header + one row + + def test_respawn_does_not_revive_a_finished_worker(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.finish("t1") + roster.spawn("t1", "hive-worker", 0) + assert roster.live_count == 0 + + def test_spawn_names_a_worker_adopted_earlier(self, roster: HiveRoster) -> None: + """A worker that spoke before its dispatch was seen gets its real name.""" + roster.note_activity("t1", "reading the module") + roster.spawn("t1", "backend-dev", 0) + assert any("backend-dev#1" in line for line in roster.render()) + + +# ============================================================================= +# note_activity - unknown ids are adopted, text is normalised +# ============================================================================= + + +class TestNoteActivity: + def test_unknown_id_is_adopted_not_dropped(self, roster: HiveRoster) -> None: + """A worker can speak before its spawning tool call is processed.""" + roster.note_activity("t9", "editing src/api/routes.py") + assert roster.live_count == 1 + assert any("editing src/api/routes.py" in line for line in roster.render()) + + def test_adopted_worker_gets_the_next_ordinal(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t9", "working") + assert any("subagent#2" in line for line in roster.render()) + + def test_newlines_collapsed(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", "running\n the\tunit tests\n") + row = next(line for line in roster.render() if "hive-worker#1" in line) + assert "running the unit tests" in row + assert "\n" not in row + + def test_long_text_truncated(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", "x" * 200) + row = next(line for line in roster.render() if "hive-worker#1" in line) + assert "…" in row + assert "x" * (MAX_ACTIVITY_CHARS + 1) not in row + + def test_exact_width_not_truncated(self, roster: HiveRoster) -> None: + text = "y" * MAX_ACTIVITY_CHARS + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", text) + assert any(text in line for line in roster.render()) + + def test_blank_ignored(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", "running the unit tests") + roster.note_activity("t1", " \n ") + assert any("running the unit tests" in line for line in roster.render()) + + def test_replaces_rather_than_accumulates(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", "reading files") + roster.note_activity("t1", "writing tests") + row = next(line for line in roster.render() if "hive-worker#1" in line) + assert "writing tests" in row + assert "reading files" not in row + + def test_non_string_does_not_raise(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", 42) # type: ignore[arg-type] + assert any("42" in line for line in roster.render()) + + +# ============================================================================= +# add_usage - both SDK shapes, garbage adds zero +# ============================================================================= + + +class _AttrUsage: + """An attribute-carrying usage payload (the older SDK shape).""" + + def __init__(self, **fields: int) -> None: + for key, value in fields.items(): + setattr(self, key, value) + + +class TestAddUsage: + def _tokens(self, roster: HiveRoster) -> str: + return next(line for line in roster.render() if "#1" in line) + + def test_none_does_not_raise(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", None) + assert "in 0 out 0" in self._tokens(roster) + + def test_garbage_values_add_zero(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": "lots", "output_tokens": None}) + assert "in 0 out 0" in self._tokens(roster) + + def test_garbage_payload_adds_zero(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", "not a usage object") + assert "in 0 out 0" in self._tokens(roster) + + def test_negative_values_add_zero(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": -5000, "output_tokens": -1}) + assert "in 0 out 0" in self._tokens(roster) + + def test_unknown_id_is_adopted(self, roster: HiveRoster) -> None: + roster.add_usage("t9", {"input_tokens": 1000, "output_tokens": 1000}) + assert roster.live_count == 1 + assert "in 1.0k out 1.0k" in self._tokens(roster) + + def test_dict_shape(self, roster: HiveRoster) -> None: + """A plain dict is what the Python SDK actually sends.""" + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": 12_000, "output_tokens": 3400}) + assert "in 12.0k out 3.4k" in self._tokens(roster) + + def test_attribute_shape(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", _AttrUsage(input_tokens=12_000, output_tokens=3400)) + assert "in 12.0k out 3.4k" in self._tokens(roster) + + def test_missing_fields_add_zero(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", _AttrUsage(output_tokens=500)) + assert "in 0 out 500" in self._tokens(roster) + + def test_cache_tokens_counted_as_input(self, roster: HiveRoster) -> None: + """A worker's cold start is mostly cache reads; omitting them lies.""" + roster.spawn("t1", "hive-worker", 0) + roster.add_usage( + "t1", + { + "input_tokens": 2000, + "cache_read_input_tokens": 400_000, + "cache_creation_input_tokens": 8000, + "output_tokens": 1000, + }, + ) + assert "in 410.0k out 1.0k" in self._tokens(roster) + + def test_cache_tokens_on_attribute_shape(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", _AttrUsage(input_tokens=1000, cache_read_input_tokens=9000)) + assert "in 10.0k out 0" in self._tokens(roster) + + def test_accumulates_across_calls(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": 1000, "output_tokens": 100}) + roster.add_usage("t1", {"input_tokens": 2000, "output_tokens": 200}) + roster.add_usage("t1", None) + roster.add_usage("t1", {"cache_read_input_tokens": 7000}) + assert "in 10.0k out 300" in self._tokens(roster) + + def test_usage_after_finish_still_counted(self, roster: HiveRoster) -> None: + """The result carrying the final usage arrives with the finish, or after.""" + roster.spawn("t1", "hive-worker", 0) + roster.finish("t1") + roster.add_usage("t1", {"input_tokens": 5000, "output_tokens": 500}) + assert "in 5.0k out 500" in self._tokens(roster) + + +# ============================================================================= +# due() - the throttle +# ============================================================================= + + +class TestDue: + def test_empty_roster_is_never_due(self, roster: HiveRoster) -> None: + """Polling before any worker exists must not spend the first window.""" + assert roster.due() is False + + def test_empty_roster_does_not_consume_the_interval( + self, roster: HiveRoster, clock: FakeClock + ) -> None: + assert roster.due() is False + clock.advance(1) + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + + def test_first_call_with_workers_is_due(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + + def test_second_call_within_interval_is_not_due(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + assert roster.due() is False + + def test_due_again_after_the_interval(self, roster: HiveRoster, clock: FakeClock) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + clock.advance(DEFAULT_ROSTER_INTERVAL_SEC - 1) + assert roster.due() is False + clock.advance(1) + assert roster.due() is True + + def test_force_ignores_the_interval(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + assert roster.due(force=True) is True + + def test_force_resets_the_timer(self, roster: HiveRoster, clock: FakeClock) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due(force=True) is True + clock.advance(DEFAULT_ROSTER_INTERVAL_SEC - 1) + assert roster.due() is False + + def test_custom_interval( + self, roster: HiveRoster, clock: FakeClock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "10") + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + clock.advance(10) + assert roster.due() is True + + def test_zero_disables_due_even_with_force( + self, roster: HiveRoster, monkeypatch: pytest.MonkeyPatch + ) -> None: + """'Off' that a caller can override is not off.""" + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "0") + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is False + assert roster.due(force=True) is False + + def test_unparseable_interval_falls_back( + self, roster: HiveRoster, clock: FakeClock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "sometimes") + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + clock.advance(DEFAULT_ROSTER_INTERVAL_SEC - 1) + assert roster.due() is False + clock.advance(1) + assert roster.due() is True + + def test_clear_resets_the_timer(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.due() is True + roster.clear() + roster.spawn("t2", "hive-worker", 0) + assert roster.due() is True + + +# ============================================================================= +# render() - plain text, aligned, empty when there is nothing to say +# ============================================================================= + + +class TestRender: + def test_empty_when_no_workers(self, roster: HiveRoster) -> None: + assert roster.render() == [] + + def test_empty_when_disabled(self, roster: HiveRoster, monkeypatch: pytest.MonkeyPatch) -> None: + roster.spawn("t1", "hive-worker", 0) + monkeypatch.setenv(ROSTER_INTERVAL_ENV, "0") + assert roster.render() == [] + + def test_no_ansi_escapes(self, roster: HiveRoster) -> None: + """Colour policy lives in console; the roster returns plain text.""" + roster.spawn("t1", "hive-worker", 0) + roster.note_activity("t1", "running the unit tests") + roster.spawn("t2", "backend-dev", 1) + roster.finish("t2", is_error=True) + assert all("\x1b" not in line for line in roster.render()) + + def test_no_emoji(self, roster: HiveRoster) -> None: + """CLAUDE.md: no emoji anywhere in the printer. ASCII marks only.""" + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "backend-dev", 1) + roster.finish("t2") + for line in roster.render(): + assert all(ord(ch) < 0x2100 for ch in line), line + + def test_header_counts_active_and_done(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "backend-dev", 1) + roster.spawn("t3", "hive-worker", 2) + roster.finish("t3") + assert roster.render()[0] == "Hive: 2 workers active, 1 done" + + def test_header_singular(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + assert roster.render()[0] == "Hive: 1 worker active" + + def test_header_all_done(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.finish("t1") + assert roster.render()[0] == "Hive: 1 done" + + def test_one_line_per_worker(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "hive-worker", 1) + assert len(roster.render()) == 3 # header + 2 rows + + def test_rows_in_ordinal_order(self, roster: HiveRoster) -> None: + roster.spawn("t2", "hive-worker", 1) + roster.spawn("t1", "backend-dev", 0) + rows = roster.render()[1:] + assert "backend-dev#1" in rows[0] + assert "hive-worker#2" in rows[1] + + def test_live_marker_and_placeholder_activity(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + row = roster.render()[1] + assert row.lstrip().startswith("~") + assert "starting" in row + + def test_columns_align_to_the_longest_label(self, roster: HiveRoster) -> None: + """A long specialist name must not wreck the other columns.""" + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "database-migration-specialist", 1) + roster.note_activity("t1", "editing one") + roster.note_activity("t2", "editing two") + rows = roster.render()[1:] + assert rows[0].index("editing one") == rows[1].index("editing two") + + def test_columns_align_across_differing_activities(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "hive-worker", 1) + roster.note_activity("t1", "a") + roster.note_activity("t2", "a much longer description of the work") + rows = roster.render()[1:] + assert rows[0].index(" in ") == rows[1].index(" in ") + + def test_no_trailing_whitespace(self, roster: HiveRoster) -> None: + roster.spawn("t1", "hive-worker", 0) + roster.add_usage("t1", {"input_tokens": 1000, "output_tokens": 100}) + assert all(line == line.rstrip() for line in roster.render()) + + def test_full_block(self, roster: HiveRoster, clock: FakeClock) -> None: + """The whole thing, as a reader would see it in a log.""" + roster.spawn("t1", "hive-worker", 0) + roster.spawn("t2", "backend-dev", 1) + roster.add_usage("t1", {"input_tokens": 410_200, "output_tokens": 8100}) + roster.add_usage("t2", {"input_tokens": 180_200, "output_tokens": 6000}) + clock.advance(62) + roster.spawn("t3", "hive-worker", 2) + roster.add_usage("t3", {"input_tokens": 40_000, "output_tokens": 2200}) + roster.finish("t3") + clock.advance(190) + roster.note_activity("t1", "running the unit tests") + roster.note_activity("t2", "editing src/api/routes.py") + assert roster.render() == [ + "Hive: 2 workers active, 1 done", + " ~ hive-worker#1 running the unit tests 4m 12s in 410.2k out 8.1k", + " ~ backend-dev#2 editing src/api/routes.py 4m 12s in 180.2k out 6.0k", + " + hive-worker#3 done 0s in 40.0k out 2.2k", + ] diff --git a/uv.lock b/uv.lock index 4a621a6a..88625ba1 100644 --- a/uv.lock +++ b/uv.lock @@ -196,20 +196,21 @@ wheels = [ [[package]] name = "claude-agent-sdk" -version = "0.2.137" +version = "0.2.150" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, + { name = "jsonschema" }, { name = "mcp" }, { name = "sniffio" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/d4/f0/6f4bc30760c4df9a7b30905bd8426ee690fc43b9893c4981c342e5a2135d/claude_agent_sdk-0.2.137.tar.gz", hash = "sha256:1fc9fdd2e3ea9674ebfe1d3580a7e44a847e4832ff02058ada41cf4af1541789", size = 319524, upload-time = "2026-08-12T21:09:55.222Z" } +sdist = { url = "https://files.pythonhosted.org/packages/6d/1e/99bababecf24a4a55a0775665ff0be010aff26f653fc9be2f4f17a0b5ef0/claude_agent_sdk-0.2.150.tar.gz", hash = "sha256:3856e932f8d45514ddf5ddd540fb844678bb11638cabf9b939ce8b708e4eb352", size = 344692, upload-time = "2026-09-01T18:07:15.284Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/28/1c/b46b79fe614ca6cdb9c509477a3fdd0576cf688019c9bedb91eeb3e77000/claude_agent_sdk-0.2.137-py3-none-macosx_11_0_arm64.whl", hash = "sha256:ae826d1d2448209919b514ee909de005ed83344c764d09cb25f44d67b004c8a9", size = 84813968, upload-time = "2026-08-12T21:09:59.246Z" }, - { url = "https://files.pythonhosted.org/packages/d8/fe/0babe13e4217e7e0c4edf0ec1a98f12068c6ccca046cfededaee8afa6a8c/claude_agent_sdk-0.2.137-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:d636fc3781e2bc4ced34db5fa7909c12644be6138cda66539c580425c9e67d80", size = 89748359, upload-time = "2026-08-12T21:10:03.204Z" }, - { url = "https://files.pythonhosted.org/packages/0f/13/7fc6a12bcb3abf1527014dc0b7ca89cd7b453faf11d4a077b71a85e5d451/claude_agent_sdk-0.2.137-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:7e85fa7d3ec66d69ecc4f283f4849b3b253cbc3d4e88990989ee1d43d1625f58", size = 93533868, upload-time = "2026-08-12T21:10:07.131Z" }, - { url = "https://files.pythonhosted.org/packages/10/be/b0e2be5f6e938d14f056926584d16061e187d465bdcf57c27ceeda7711fb/claude_agent_sdk-0.2.137-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:b75d771e7e076d3b64b6e7b3e8583250505d516376b21327ce60a6af1f56a3aa", size = 94571140, upload-time = "2026-08-12T21:10:11.463Z" }, - { url = "https://files.pythonhosted.org/packages/1f/52/df5460e4a9f4d66ba6d447350f7510052fc0578f5995548efcd8cd5cf2a0/claude_agent_sdk-0.2.137-py3-none-win_amd64.whl", hash = "sha256:596ee0f8b3411bd8061838bbc16bba60ecc6c8b6b936369377bba5231026ac8f", size = 96883369, upload-time = "2026-08-12T21:10:15.572Z" }, + { url = "https://files.pythonhosted.org/packages/c3/13/b5b86d90c95e8a691143eeb45bbf2e1c250324817980464d6ea63ef0f789/claude_agent_sdk-0.2.150-py3-none-macosx_11_0_arm64.whl", hash = "sha256:8f58d746f3552faeac52883b5a502a8d9b5aef067ea0b68fae5c9bae0f39d73d", size = 85396845, upload-time = "2026-09-01T18:07:19.074Z" }, + { url = "https://files.pythonhosted.org/packages/d9/8f/c17da4fa9cd8bddb8ac1538456dd0fde3a43abb1dfc28ca448f85e2a10d3/claude_agent_sdk-0.2.150-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:fee3ae2dd457742d82070b5fa122070ee8e2e80b4f9b37a57ff0a8261ade656d", size = 89904483, upload-time = "2026-09-01T18:07:22.671Z" }, + { url = "https://files.pythonhosted.org/packages/8d/95/5b33717b807c6fbe29fca55a246eab9a09a80c9dd288ffeb31f05681b7fc/claude_agent_sdk-0.2.150-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:813535a0267271d91af29c456ec26137b34a8b7bc066647ea59efba6e3c26c93", size = 95270310, upload-time = "2026-09-01T18:07:26.61Z" }, + { url = "https://files.pythonhosted.org/packages/9f/de/5a8b59e6a713ea3f4be88816cfd487b1a5b08c8ad9f042abe56d2decc184/claude_agent_sdk-0.2.150-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:cb65ff33060baa098b420232c630fa287d0f95e9469e7ceb706d0b7d8841888e", size = 95460277, upload-time = "2026-09-01T18:07:31.092Z" }, + { url = "https://files.pythonhosted.org/packages/dd/42/d147578a0df16c1b4b0deaebfb3d25e590ad5c16cb6b59a68f9dbb618be3/claude_agent_sdk-0.2.150-py3-none-win_amd64.whl", hash = "sha256:3b517a21c723854de778b38fa51660c5b68ee5fb6f28b2abf291a51471a7b378", size = 98078644, upload-time = "2026-09-01T18:07:35.734Z" }, ] [[package]] @@ -267,13 +268,13 @@ mcp = [ [package.metadata] requires-dist = [ { name = "bcrypt", marker = "extra == 'api'", specifier = ">=4.0.0,<5.0.0" }, - { name = "claude-agent-sdk", specifier = ">=0.2.137,<0.3" }, + { name = "claude-agent-sdk", specifier = ">=0.2.150,<0.3" }, { name = "claude-task-master", extras = ["api"], marker = "extra == 'dev'" }, { name = "claude-task-master", extras = ["mcp", "api", "dev"], marker = "extra == 'all'" }, { name = "fastapi", marker = "extra == 'api'", specifier = ">=0.128.7" }, { name = "httpx", specifier = ">=0.28.0,<1" }, { name = "hypothesis", marker = "extra == 'dev'", specifier = ">=6.151.6" }, - { name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.26.0" }, + { name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.26.0,<2" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.19.0,<2" }, { name = "passlib", extras = ["bcrypt"], marker = "extra == 'api'", specifier = ">=1.7.4" }, { name = "pydantic", specifier = ">=2.12.0,<3" }, From 4886c20e89598c525da6360e9207374a1cb05330 Mon Sep 17 00:00:00 2001 From: sebi Date: Tue, 1 Sep 2026 16:21:32 -0500 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20address=20review=20=E2=80=94=20bare?= =?UTF-8?q?=20status=20numbers,=20stale=20README=20default,=20long=20line?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three of CodeRabbit's five findings were real. A bare 401/403 in echoed command output classified as a non-retryable APIAuthenticationError. This is the same trap the PR set out to fix, one step further in: folding the result/errors payload into the searched text is what lets a real failure be recognised, and it drags tool output in with it, so "wrote 403 bytes" and "exit 401" matched. Auth status matching now needs a status-like word nearby; 5xx deliberately does not, because a false 5xx is retried while a false auth error ends the run. "Forbidden" is matched as a word so "403 Forbidden" survives regardless. README documented CLAUDETM_MAX_TURNS as 400; the code and CLAUDE.md say 2000. Stale since the hive sizing change. Split a 140-char source line (repo limit is 100) with a continuation, so the rendered prompt is byte-identical — asserted before and after. Not taken: the roster leaving successful workers "active" is deliberate and documented — a dispatch's tool result is an ACK, and the stream carries no correlated terminal event for a worker (see #160 for why the first ResultMessage cannot stand in for one). Test-naming nit skipped; the class name already carries the function under test, per repo style. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Phc3xittodp1pz7ZeMRAfa --- CHANGELOG.md | 1 + README.md | 2 +- .../core/agent_error_classify.py | 22 +++++++-- .../core/prompts_working_hive.py | 4 +- tests/core/test_agent_error_classify.py | 49 +++++++++++++++++++ 5 files changed, 73 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 26c85726..fd7ea429 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `str(ResultError)` is **only** `"Claude Code returned an error result: (exit code: 1)"` — the prose naming the actual failure lives in `errors`/`result`. Folding the payload into the searched text is what makes **529 overloaded** a retryable `APIServerError` (it appears nowhere in `str()`, so the old rules classified it as an unrecognised error) and what lets `Connection closed mid-response` — the blip that once ended a 22-task unattended run at task 1 — classify as `APIConnectionError` at all. +- **A bare `401`/`403` in echoed command output is no longer an auth error.** Folding the `result`/`errors` payload into the searched text is what lets a real failure be recognised, and it drags tool and command output in with it — so `wrote 403 bytes` or `exit 401` matched a digit-boundary rule. Auth status matching now requires a status-like word nearby; the 5xx rule deliberately does not need to, because a false 5xx is merely retried while a false auth error is not retryable and ends the run. `Forbidden` is also matched as a word, so `403 Forbidden` survives without the digits. - **Two loose substring rules, each wrong in one direction.** `"500" in text` also matched `request took 1500ms`, promoting a latency message to a retryable `APIServerError`; `"auth" in text` also matched **`Co-Authored-By`**, which appears in every commit message this project writes, so any error echoing a git command became a *non*-retryable `APIAuthenticationError` that ended the run. Status codes and auth phrasings now match on token boundaries, pinned by named regression tests. ## [0.1.89] - 2026-08-14 diff --git a/README.md b/README.md index b445a89b..2874f4f1 100644 --- a/README.md +++ b/README.md @@ -380,7 +380,7 @@ Precedence (highest first): real environment variables, then the **active profil | Environment Variable | Default | Description | |---------------------|---------|-------------| -| `CLAUDETM_MAX_TURNS` | `400` | Max agent steps per session — a runaway backstop, not a working budget. Set `0` to disable. Overrunning retries the task rather than marking it done | +| `CLAUDETM_MAX_TURNS` | `2000` | Max agent steps per session — a runaway backstop, not a working budget. Counts the lead **and every subagent**, so it is sized for a hive. Set `0` to disable. Overrunning retries the task rather than marking it done | | `CLAUDETM_STREAM_IDLE_TIMEOUT_SEC` | `1800` | Max silence between SDK stream messages before treating the stream as hung | | `CLAUDETM_POST_COMPLETION_IDLE_TIMEOUT_SEC` | `120` | Max wait for the final result message after the agent signals it's done | | `CLAUDETM_HIVE_MAX_PARALLEL` | `6` | Safety ceiling on concurrent worker subagents (1 lead + up to N workers) — a ceiling, not a target; the lead sizes its own team, and may run N of the same kind, N different kinds, or any mix | diff --git a/src/claude_task_master/core/agent_error_classify.py b/src/claude_task_master/core/agent_error_classify.py index 2ec0a0b8..8d4736e1 100644 --- a/src/claude_task_master/core/agent_error_classify.py +++ b/src/claude_task_master/core/agent_error_classify.py @@ -32,6 +32,12 @@ commit message this project writes — turning any error that echoed a git command into a *non*-retryable :class:`APIAuthenticationError` that ends the run. + +A bare ``401``/``403`` is the same trap one step further in: because the payload +prose is searched, echoed command output such as ``"wrote 403 bytes"`` would +match. Status matching for auth therefore requires a status-like word nearby, +while the 5xx rule does not need to — a false 5xx is retried, a false auth kills +the run. """ from __future__ import annotations @@ -56,13 +62,23 @@ #: A standalone HTTP 5xx status, not a digit inside a larger number. _SERVER_STATUS_RE = re.compile(r"(? None: + assert not isinstance(classify_api_error(Exception(message)), APIAuthenticationError) + + @pytest.mark.parametrize( + "message", + [ + "HTTP 401 Unauthorized", + "HTTP 403 Forbidden", + "status: 403", + "error 401 returned by the API", + "response code 403", + ], + ) + def test_status_with_context_is_still_an_auth_error(self, message: str) -> None: + assert isinstance(classify_api_error(Exception(message)), APIAuthenticationError) + + def test_bare_number_inside_result_payload_prose(self) -> None: + """The realistic shape: the number arrives via `errors`, not `str(e)`.""" + err = Exception("Claude Code returned an error result: error_during_execution") + err.errors = ["Command output: wrote 403 bytes"] # type: ignore[attr-defined] + assert not isinstance(classify_api_error(err), APIAuthenticationError) + + def test_forbidden_word_alone_is_an_auth_error(self) -> None: + """`403 Forbidden` must survive even without the digits.""" + assert isinstance( + classify_api_error(Exception("Forbidden: you lack access")), + APIAuthenticationError, + )