12 Commits

Author SHA1 Message Date
Yeachan-Heo
61b4def7bc feat: telemetry progress 2026-04-01 06:15:15 +00:00
Yeachan-Heo
1b42c6096c feat: anthropic SDK header matching + request profile 2026-04-01 05:55:25 +00:00
Yeachan-Heo
828597024e wip: telemetry claude code matching 2026-04-01 05:45:28 +00:00
Yeachan-Heo
e7e3ae2875 wip: telemetry progress 2026-04-01 04:40:21 +00:00
Yeachan-Heo
5170718306 wip: telemetry progress 2026-04-01 04:30:29 +00:00
Yeachan-Heo
ac6c5d00a8 Enable Claude-compatible tool hooks in the Rust runtime
This threads typed hook settings through runtime config, adds a shell-based hook runner, and executes PreToolUse/PostToolUse around each tool call in the conversation loop. The CLI now rebuilds runtimes with settings-derived hook configuration so user-defined Claude hook commands actually run before and after tools.

Constraint: Hook behavior needed to match Claude-style settings.json hooks without broad plugin/MCP parity work in this change
Rejected: Delay hook loading to the tool executor layer | would miss denied tool calls and duplicate runtime policy plumbing
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Keep hook execution in the runtime loop so permission decisions and tool results remain wrapped by the same conversation semantics
Tested: cargo test; cargo build --release
Not-tested: Real user hook scripts outside the test harness; broader plugin/skills parity
2026-04-01 03:35:25 +00:00
Yeachan-Heo
a94ef61b01 feat: -p flag compat, --print flag, OAuth defaults, UI rendering merge 2026-04-01 03:22:34 +00:00
Yeachan-Heo
a9ac7e5bb8 feat: default OAuth config for claude.com, merge UI polish rendering 2026-04-01 03:20:26 +00:00
Yeachan-Heo
0175ee0a90 Merge remote-tracking branch 'origin/rcc/ui-polish' into dev/rust 2026-04-01 03:17:16 +00:00
Yeachan-Heo
1bd0eef368 Merge remote-tracking branch 'origin/rcc/subagent' into dev/rust 2026-04-01 03:12:25 +00:00
Yeachan-Heo
ba220d210e Enable real Agent tool delegation in the Rust CLI
The Rust Agent tool only persisted queued metadata, so delegated work never actually ran. This change wires Agent into a detached background conversation path with isolated runtime, API client, session state, restricted tool subsets, and file-backed lifecycle/result updates.

Constraint: Keep the tool entrypoint in the tools crate and avoid copying the upstream TypeScript implementation
Rejected: Spawn an external claw process | less aligned with the requested in-process runtime/client design
Rejected: Leave execution in the CLI crate only | would keep tools::Agent as a metadata-only stub
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Tool subset mappings are curated guardrails; revisit them before enabling recursive Agent access or richer agent definitions
Tested: cargo build --release --manifest-path rust/Cargo.toml
Tested: cargo test --manifest-path rust/Cargo.toml
Not-tested: Live end-to-end background sub-agent run against Anthropic API credentials
2026-04-01 03:10:20 +00:00
Yeachan-Heo
04b1f1e85d docs: rewrite rust/ README with full feature matrix and usage guide 2026-04-01 02:59:05 +00:00
26 changed files with 2932 additions and 336 deletions

5
.claude.json Normal file
View File

@@ -0,0 +1,5 @@
{
"permissions": {
"defaultMode": "dontAsk"
}
}

3
.gitignore vendored
View File

@@ -2,3 +2,6 @@ __pycache__/
archive/ archive/
.omx/ .omx/
.clawd-agents/ .clawd-agents/
# Claude Code local artifacts
.claude/settings.local.json
.claude/sessions/

21
CLAUDE.md Normal file
View File

@@ -0,0 +1,21 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Detected stack
- Languages: Rust.
- Frameworks: none detected from the supported starter markers.
## Verification
- Run Rust verification from `rust/`: `cargo fmt`, `cargo clippy --workspace --all-targets -- -D warnings`, `cargo test --workspace`
- `src/` and `tests/` are both present; update both surfaces together when behavior changes.
## Repository shape
- `rust/` contains the Rust workspace and active CLI/runtime implementation.
- `src/` contains source files that should stay consistent with generated guidance and tests.
- `tests/` contains validation surfaces that should be reviewed alongside code changes.
## Working agreement
- Prefer small, reviewable changes and keep generated bootstrap files aligned with actual repo workflows.
- Keep shared defaults in `.claude.json`; reserve `.claude/settings.local.json` for machine-local overrides.
- Do not overwrite existing `CLAUDE.md` content automatically; update it intentionally when repo workflows change.

View File

@@ -0,0 +1 @@
{"messages":[{"blocks":[{"text":"clear","type":"text"}],"role":"user"},{"blocks":[{"text":"\n\nI've cleared the conversation. How can I help you today?","type":"text"}],"role":"assistant","usage":{"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"input_tokens":4272,"output_tokens":17}}],"version":1}

View File

@@ -0,0 +1 @@
{"messages":[{"blocks":[{"text":"exit","type":"text"}],"role":"user"},{"blocks":[{"text":"\n\nGoodbye! 👋","type":"text"}],"role":"assistant","usage":{"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"input_tokens":4272,"output_tokens":10}}],"version":1}

View File

@@ -0,0 +1 @@
{"messages":[],"version":1}

View File

@@ -1,22 +1,27 @@
[ [
{ {
"content": "Phase 0: Structural Cleanup — spawn 4 agents for 0.1-0.4", "content": "Architecture & dependency analysis",
"activeForm": "Executing Phase 0: Structural Cleanup via sub-agents", "activeForm": "Complete",
"status": "completed"
},
{
"content": "Runtime crate deep analysis",
"activeForm": "Complete",
"status": "completed"
},
{
"content": "CLI & Tools analysis",
"activeForm": "Complete",
"status": "completed"
},
{
"content": "Code quality verification",
"activeForm": "Complete",
"status": "completed"
},
{
"content": "Synthesize findings into unified report",
"activeForm": "Writing report",
"status": "in_progress" "status": "in_progress"
},
{
"content": "Phase 1.1-1.2: Status bar with live HUD and token counter",
"activeForm": "Awaiting Phase 0",
"status": "pending"
},
{
"content": "Phase 2.4: Remove artificial 8ms stream delay",
"activeForm": "Awaiting Phase 0",
"status": "pending"
},
{
"content": "Phase 3.1: Collapsible tool output",
"activeForm": "Awaiting Phase 0",
"status": "pending"
} }
] ]

12
rust/Cargo.lock generated
View File

@@ -25,6 +25,7 @@ dependencies = [
"runtime", "runtime",
"serde", "serde",
"serde_json", "serde_json",
"telemetry",
"tokio", "tokio",
] ]
@@ -1096,6 +1097,7 @@ dependencies = [
"serde", "serde",
"serde_json", "serde_json",
"sha2", "sha2",
"telemetry",
"tokio", "tokio",
"walkdir", "walkdir",
] ]
@@ -1428,6 +1430,14 @@ dependencies = [
"yaml-rust", "yaml-rust",
] ]
[[package]]
name = "telemetry"
version = "0.1.0"
dependencies = [
"serde",
"serde_json",
]
[[package]] [[package]]
name = "thiserror" name = "thiserror"
version = "2.0.18" version = "2.0.18"
@@ -1545,10 +1555,12 @@ dependencies = [
name = "tools" name = "tools"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"api",
"reqwest", "reqwest",
"runtime", "runtime",
"serde", "serde",
"serde_json", "serde_json",
"tokio",
] ]
[[package]] [[package]]

View File

@@ -1,230 +1,149 @@
# Rusty Claude CLI # 🦞 Claw Code — Rust Implementation
`rust/` contains the Rust workspace for the integrated `rusty-claude-cli` deliverable. A high-performance Rust rewrite of the Claude Code CLI agent harness. Built for speed, safety, and native tool execution.
It is intended to be something you can clone, build, and run directly.
## Workspace layout ## Quick Start
```text ```bash
# Build
cd rust/
cargo build --release
# Run interactive REPL
./target/release/claw
# One-shot prompt
./target/release/claw prompt "explain this codebase"
# With specific model
./target/release/claw --model sonnet prompt "fix the bug in main.rs"
```
## Configuration
Set your API credentials:
```bash
export ANTHROPIC_API_KEY="sk-ant-..."
# Or use a proxy
export ANTHROPIC_BASE_URL="https://your-proxy.com"
```
Or authenticate via OAuth:
```bash
claw login
```
## Features
| Feature | Status |
|---------|--------|
| Anthropic API + streaming | ✅ |
| OAuth login/logout | ✅ |
| Interactive REPL (rustyline) | ✅ |
| Tool system (bash, read, write, edit, grep, glob) | ✅ |
| Web tools (search, fetch) | ✅ |
| Sub-agent orchestration | ✅ |
| Todo tracking | ✅ |
| Notebook editing | ✅ |
| CLAUDE.md / project memory | ✅ |
| Config file hierarchy (.claude.json) | ✅ |
| Permission system | ✅ |
| MCP server lifecycle | ✅ |
| Session persistence + resume | ✅ |
| Extended thinking (thinking blocks) | ✅ |
| Cost tracking + usage display | ✅ |
| Git integration | ✅ |
| Markdown terminal rendering (ANSI) | ✅ |
| Model aliases (opus/sonnet/haiku) | ✅ |
| Slash commands (/status, /compact, /clear, etc.) | ✅ |
| Hooks (PreToolUse/PostToolUse) | 🔧 Config only |
| Plugin system | 📋 Planned |
| Skills registry | 📋 Planned |
## Model Aliases
Short names resolve to the latest model versions:
| Alias | Resolves To |
|-------|------------|
| `opus` | `claude-opus-4-6` |
| `sonnet` | `claude-sonnet-4-6` |
| `haiku` | `claude-haiku-4-5-20251213` |
## CLI Flags
```
claw [OPTIONS] [COMMAND]
Options:
--model MODEL Set the model (alias or full name)
--dangerously-skip-permissions Skip all permission checks
--permission-mode MODE Set read-only, workspace-write, or danger-full-access
--allowedTools TOOLS Restrict enabled tools
--output-format FORMAT Output format (text or json)
--version, -V Print version info
Commands:
prompt <text> One-shot prompt (non-interactive)
login Authenticate via OAuth
logout Clear stored credentials
init Initialize project config
doctor Check environment health
self-update Update to latest version
```
## Slash Commands (REPL)
| Command | Description |
|---------|-------------|
| `/help` | Show help |
| `/status` | Show session status (model, tokens, cost) |
| `/cost` | Show cost breakdown |
| `/compact` | Compact conversation history |
| `/clear` | Clear conversation |
| `/model [name]` | Show or switch model |
| `/permissions` | Show or switch permission mode |
| `/config [section]` | Show config (env, hooks, model) |
| `/memory` | Show CLAUDE.md contents |
| `/diff` | Show git diff |
| `/export [path]` | Export conversation |
| `/session [id]` | Resume a previous session |
| `/version` | Show version |
## Workspace Layout
```
rust/ rust/
├── Cargo.toml ├── Cargo.toml # Workspace root
├── Cargo.lock ├── Cargo.lock
├── README.md
└── crates/ └── crates/
├── api/ # Anthropic API client + SSE streaming support ├── api/ # Anthropic API client + SSE streaming
├── commands/ # Shared slash-command metadata/help surfaces ├── commands/ # Shared slash-command registry
├── compat-harness/ # Upstream TS manifest extraction harness ├── compat-harness/ # TS manifest extraction harness
├── runtime/ # Session/runtime/config/prompt orchestration ├── runtime/ # Session, config, permissions, MCP, prompts
├── rusty-claude-cli/ # Main CLI binary ├── rusty-claude-cli/ # Main CLI binary (`claw`)
└── tools/ # Built-in tool implementations └── tools/ # Built-in tool implementations
``` ```
## Prerequisites ### Crate Responsibilities
- Rust toolchain installed (`rustup`, stable toolchain) - **api** — HTTP client, SSE stream parser, request/response types, auth (API key + OAuth bearer)
- Network access and Anthropic credentials for live prompt/REPL usage - **commands** — Slash command definitions and help text generation
- **compat-harness** — Extracts tool/prompt manifests from upstream TS source
- **runtime** — `ConversationRuntime` agentic loop, `ConfigLoader` hierarchy, `Session` persistence, permission policy, MCP client, system prompt assembly, usage tracking
- **rusty-claude-cli** — REPL, one-shot prompt, streaming display, tool call rendering, CLI argument parsing
- **tools** — Tool specs + execution: Bash, ReadFile, WriteFile, EditFile, GlobSearch, GrepSearch, WebSearch, WebFetch, Agent, TodoWrite, NotebookEdit, Skill, ToolSearch, REPL runtimes
## Build ## Stats
From the repository root: - **~20K lines** of Rust
- **6 crates** in workspace
- **Binary name:** `claw`
- **Default model:** `claude-opus-4-6`
- **Default permissions:** `danger-full-access`
```bash ## License
cd rust
cargo build --release -p rusty-claude-cli
```
The optimized binary will be written to: See repository root.
```bash
./target/release/rusty-claude-cli
```
## Test
Run the verified workspace test suite used for release-readiness:
```bash
cd rust
cargo test --workspace --exclude compat-harness
```
## Quick start
### Show help
```bash
cd rust
cargo run -p rusty-claude-cli -- --help
```
### Print version
```bash
cd rust
cargo run -p rusty-claude-cli -- --version
```
### Login with OAuth
Configure `settings.json` with an `oauth` block containing `clientId`, `authorizeUrl`, `tokenUrl`, optional `callbackPort`, and optional `scopes`, then run:
```bash
cd rust
cargo run -p rusty-claude-cli -- login
```
This opens the browser, listens on the configured localhost callback, exchanges the auth code for tokens, and stores OAuth credentials in `~/.claude/credentials.json` (or `$CLAUDE_CONFIG_HOME/credentials.json`).
### Logout
```bash
cd rust
cargo run -p rusty-claude-cli -- logout
```
This removes only the stored OAuth credentials and preserves unrelated JSON fields in `credentials.json`.
### Self-update
```bash
cd rust
cargo run -p rusty-claude-cli -- self-update
```
The command checks the latest GitHub release for `instructkr/clawd-code`, compares it to the current binary version, downloads the matching binary asset plus checksum manifest, verifies SHA-256, replaces the current executable, and prints the release changelog. If no published release or matching asset exists, it exits safely with an explanatory message.
## Usage examples
### 1) Prompt mode
Send one prompt, stream the answer, then exit:
```bash
cd rust
cargo run -p rusty-claude-cli -- prompt "Summarize the architecture of this repository"
```
Use a specific model:
```bash
cd rust
cargo run -p rusty-claude-cli -- --model claude-sonnet-4-20250514 prompt "List the key crates in this workspace"
```
Restrict enabled tools in an interactive session:
```bash
cd rust
cargo run -p rusty-claude-cli -- --allowedTools read,glob
```
Bootstrap Claude project files for the current repo:
```bash
cd rust
cargo run -p rusty-claude-cli -- init
```
### 2) REPL mode
Start the interactive shell:
```bash
cd rust
cargo run -p rusty-claude-cli --
```
Inside the REPL, useful commands include:
```text
/help
/status
/model claude-sonnet-4-20250514
/permissions workspace-write
/cost
/compact
/memory
/config
/init
/diff
/version
/export notes.txt
/sessions
/session list
/exit
```
### 3) Resume an existing session
Inspect or maintain a saved session file without entering the REPL:
```bash
cd rust
cargo run -p rusty-claude-cli -- --resume session-123456 /status /compact /cost
```
You can also inspect memory/config state for a restored session:
```bash
cd rust
cargo run -p rusty-claude-cli -- --resume ~/.claude/sessions/session-123456.json /memory /config
```
## Available commands
### Top-level CLI commands
- `prompt <text...>` — run one prompt non-interactively
- `--resume <session-id-or-path> [/commands...]` — inspect or maintain a saved session stored under `~/.claude/sessions/`
- `dump-manifests` — print extracted upstream manifest counts
- `bootstrap-plan` — print the current bootstrap skeleton
- `system-prompt [--cwd PATH] [--date YYYY-MM-DD]` — render the synthesized system prompt
- `self-update` — update the installed binary from the latest GitHub release when a matching asset is available
- `--help` / `-h` — show CLI help
- `--version` / `-V` — print the CLI version and build info locally (no API call)
- `--output-format text|json` — choose non-interactive prompt output rendering
- `--allowedTools <tool[,tool...]>` — restrict enabled tools for interactive sessions and prompt-mode tool use
### Interactive slash commands
- `/help` — show command help
- `/status` — show current session status
- `/compact` — compact local session history
- `/model [model]` — inspect or switch the active model
- `/permissions [read-only|workspace-write|danger-full-access]` — inspect or switch permissions
- `/clear [--confirm]` — clear the current local session
- `/cost` — show token usage totals
- `/resume <session-id-or-path>` — load a saved session into the REPL
- `/config [env|hooks|model]` — inspect discovered Claude config
- `/memory` — inspect loaded instruction memory files
- `/init` — bootstrap `.claude.json`, `.claude/`, `CLAUDE.md`, and local ignore rules
- `/diff` — show the current git diff for the workspace
- `/version` — print version and build metadata locally
- `/export [file]` — export the current conversation transcript
- `/sessions` — list recent managed local sessions from `~/.claude/sessions/`
- `/session [list|switch <session-id>]` — inspect or switch managed local sessions
- `/exit` — leave the REPL
## Environment variables
### Anthropic/API
- `ANTHROPIC_API_KEY` — highest-precedence API credential
- `ANTHROPIC_AUTH_TOKEN` — bearer-token override used when no API key is set
- Persisted OAuth credentials in `~/.claude/credentials.json` — used when neither env var is set
- `ANTHROPIC_BASE_URL` — override the Anthropic API base URL
- `ANTHROPIC_MODEL` — default model used by selected live integration tests
### CLI/runtime
- `RUSTY_CLAUDE_PERMISSION_MODE` — default REPL permission mode (`read-only`, `workspace-write`, or `danger-full-access`)
- `CLAUDE_CONFIG_HOME` — override Claude config discovery root
- `CLAUDE_CODE_REMOTE` — enable remote-session bootstrap handling when supported
- `CLAUDE_CODE_REMOTE_SESSION_ID` — remote session identifier when using remote mode
- `CLAUDE_CODE_UPSTREAM` — override the upstream TS source path for compat-harness extraction
- `CLAWD_WEB_SEARCH_BASE_URL` — override the built-in web search service endpoint used by tooling
## Notes
- `compat-harness` exists to compare the Rust port against the upstream TypeScript codebase and is intentionally excluded from the requested release test run.
- The CLI currently focuses on a practical integrated workflow: prompt execution, REPL operation, session inspection/resume, config discovery, and tool/runtime plumbing.

View File

@@ -10,6 +10,7 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
runtime = { path = "../runtime" } runtime = { path = "../runtime" }
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
telemetry = { path = "../telemetry" }
tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "time"] } tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "time"] }
[lints] [lints]

View File

@@ -2,17 +2,19 @@ use std::collections::VecDeque;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use runtime::{ use runtime::{
load_oauth_credentials, save_oauth_credentials, OAuthConfig, OAuthRefreshRequest, format_usd, load_oauth_credentials, pricing_for_model, save_oauth_credentials, OAuthConfig,
OAuthTokenExchangeRequest, OAuthRefreshRequest, OAuthTokenExchangeRequest,
}; };
use serde::Deserialize; use serde::Deserialize;
use serde_json::{Map, Value};
use telemetry::{AnalyticsEvent, AnthropicRequestProfile, ClientIdentity, SessionTracer};
use crate::error::ApiError; use crate::error::ApiError;
use crate::sse::SseParser; use crate::sse::SseParser;
use crate::types::{MessageRequest, MessageResponse, StreamEvent}; use crate::types::{MessageRequest, MessageResponse, StreamEvent};
const DEFAULT_BASE_URL: &str = "https://api.anthropic.com"; const DEFAULT_BASE_URL: &str = "https://api.anthropic.com";
const ANTHROPIC_VERSION: &str = "2023-06-01"; const MESSAGES_PATH: &str = "/v1/messages";
const REQUEST_ID_HEADER: &str = "request-id"; const REQUEST_ID_HEADER: &str = "request-id";
const ALT_REQUEST_ID_HEADER: &str = "x-request-id"; const ALT_REQUEST_ID_HEADER: &str = "x-request-id";
const DEFAULT_INITIAL_BACKOFF: Duration = Duration::from_millis(200); const DEFAULT_INITIAL_BACKOFF: Duration = Duration::from_millis(200);
@@ -108,6 +110,8 @@ pub struct AnthropicClient {
max_retries: u32, max_retries: u32,
initial_backoff: Duration, initial_backoff: Duration,
max_backoff: Duration, max_backoff: Duration,
request_profile: AnthropicRequestProfile,
session_tracer: Option<SessionTracer>,
} }
impl AnthropicClient { impl AnthropicClient {
@@ -120,6 +124,8 @@ impl AnthropicClient {
max_retries: DEFAULT_MAX_RETRIES, max_retries: DEFAULT_MAX_RETRIES,
initial_backoff: DEFAULT_INITIAL_BACKOFF, initial_backoff: DEFAULT_INITIAL_BACKOFF,
max_backoff: DEFAULT_MAX_BACKOFF, max_backoff: DEFAULT_MAX_BACKOFF,
request_profile: AnthropicRequestProfile::default(),
session_tracer: None,
} }
} }
@@ -132,6 +138,8 @@ impl AnthropicClient {
max_retries: DEFAULT_MAX_RETRIES, max_retries: DEFAULT_MAX_RETRIES,
initial_backoff: DEFAULT_INITIAL_BACKOFF, initial_backoff: DEFAULT_INITIAL_BACKOFF,
max_backoff: DEFAULT_MAX_BACKOFF, max_backoff: DEFAULT_MAX_BACKOFF,
request_profile: AnthropicRequestProfile::default(),
session_tracer: None,
} }
} }
@@ -176,6 +184,39 @@ impl AnthropicClient {
self self
} }
#[must_use]
pub fn with_request_profile(mut self, request_profile: AnthropicRequestProfile) -> Self {
self.request_profile = request_profile;
self
}
#[must_use]
pub fn with_client_identity(mut self, client_identity: ClientIdentity) -> Self {
self.request_profile.client_identity = client_identity;
self
}
#[must_use]
pub fn with_beta(mut self, beta: impl Into<String>) -> Self {
let beta = beta.into();
if !self.request_profile.betas.contains(&beta) {
self.request_profile.betas.push(beta);
}
self
}
#[must_use]
pub fn with_extra_body_param(mut self, key: impl Into<String>, value: Value) -> Self {
self.request_profile.extra_body.insert(key.into(), value);
self
}
#[must_use]
pub fn with_session_tracer(mut self, session_tracer: SessionTracer) -> Self {
self.session_tracer = Some(session_tracer);
self
}
#[must_use] #[must_use]
pub fn with_retry_policy( pub fn with_retry_policy(
mut self, mut self,
@@ -211,6 +252,7 @@ impl AnthropicClient {
if response.request_id.is_none() { if response.request_id.is_none() {
response.request_id = request_id; response.request_id = request_id;
} }
self.record_response_usage(&response);
Ok(response) Ok(response)
} }
@@ -279,18 +321,30 @@ impl AnthropicClient {
loop { loop {
attempts += 1; attempts += 1;
self.record_request_started(request, attempts);
match self.send_raw_request(request).await { match self.send_raw_request(request).await {
Ok(response) => match expect_success(response).await { Ok(response) => match expect_success(response).await {
Ok(response) => return Ok(response), Ok(response) => {
self.record_request_succeeded(request, attempts, &response);
return Ok(response);
}
Err(error) if error.is_retryable() && attempts <= self.max_retries + 1 => { Err(error) if error.is_retryable() && attempts <= self.max_retries + 1 => {
self.record_request_failed(request, attempts, &error);
last_error = Some(error); last_error = Some(error);
} }
Err(error) => return Err(error), Err(error) => {
self.record_request_failed(request, attempts, &error);
return Err(error);
}
}, },
Err(error) if error.is_retryable() && attempts <= self.max_retries + 1 => { Err(error) if error.is_retryable() && attempts <= self.max_retries + 1 => {
self.record_request_failed(request, attempts, &error);
last_error = Some(error); last_error = Some(error);
} }
Err(error) => return Err(error), Err(error) => {
self.record_request_failed(request, attempts, &error);
return Err(error);
}
} }
if attempts > self.max_retries { if attempts > self.max_retries {
@@ -310,18 +364,213 @@ impl AnthropicClient {
&self, &self,
request: &MessageRequest, request: &MessageRequest,
) -> Result<reqwest::Response, ApiError> { ) -> Result<reqwest::Response, ApiError> {
let request_url = format!("{}/v1/messages", self.base_url.trim_end_matches('/')); let request_url = format!("{}{}", self.base_url.trim_end_matches('/'), MESSAGES_PATH);
let request_builder = self let mut request_builder = self
.http .http
.post(&request_url) .post(&request_url)
.header("anthropic-version", ANTHROPIC_VERSION)
.header("content-type", "application/json"); .header("content-type", "application/json");
for (name, value) in self.request_profile.header_pairs() {
request_builder = request_builder.header(name, value);
}
let mut request_builder = self.auth.apply(request_builder); let mut request_builder = self.auth.apply(request_builder);
request_builder = request_builder.json(request); let request_body = self.request_profile.render_json_body(request)?;
request_builder = request_builder.json(&request_body);
request_builder.send().await.map_err(ApiError::from) request_builder.send().await.map_err(ApiError::from)
} }
fn record_request_started(&self, request: &MessageRequest, attempt: u32) {
if let Some(tracer) = &self.session_tracer {
tracer.record_http_request_started(
attempt,
"POST",
MESSAGES_PATH,
self.request_attributes(request),
);
}
}
fn record_request_succeeded(
&self,
request: &MessageRequest,
attempt: u32,
response: &reqwest::Response,
) {
if let Some(tracer) = &self.session_tracer {
tracer.record_http_request_succeeded(
attempt,
"POST",
MESSAGES_PATH,
response.status().as_u16(),
request_id_from_headers(response.headers()),
self.request_attributes(request),
);
}
}
fn record_request_failed(&self, request: &MessageRequest, attempt: u32, error: &ApiError) {
if let Some(tracer) = &self.session_tracer {
tracer.record_http_request_failed(
attempt,
"POST",
MESSAGES_PATH,
error.to_string(),
error.is_retryable(),
self.error_attributes(request, error),
);
}
}
fn record_response_usage(&self, response: &MessageResponse) {
let Some(tracer) = &self.session_tracer else {
return;
};
let cost = response.usage.estimated_cost_usd(&response.model);
let pricing_source = if pricing_for_model(&response.model).is_some() {
"model-specific"
} else {
"default-sonnet"
};
let mut properties = Map::new();
properties.insert("model".to_string(), Value::String(response.model.clone()));
properties.insert(
"pricing_source".to_string(),
Value::String(pricing_source.to_string()),
);
properties.insert(
"input_tokens".to_string(),
Value::from(response.usage.input_tokens),
);
properties.insert(
"output_tokens".to_string(),
Value::from(response.usage.output_tokens),
);
properties.insert(
"cache_creation_input_tokens".to_string(),
Value::from(response.usage.cache_creation_input_tokens),
);
properties.insert(
"cache_read_input_tokens".to_string(),
Value::from(response.usage.cache_read_input_tokens),
);
properties.insert(
"total_tokens".to_string(),
Value::from(response.usage.total_tokens()),
);
properties.insert(
"estimated_cost_usd".to_string(),
Value::String(format_usd(cost.total_cost_usd())),
);
properties.insert(
"estimated_input_cost_usd".to_string(),
Value::String(format_usd(cost.input_cost_usd)),
);
properties.insert(
"estimated_output_cost_usd".to_string(),
Value::String(format_usd(cost.output_cost_usd)),
);
properties.insert(
"estimated_cache_creation_cost_usd".to_string(),
Value::String(format_usd(cost.cache_creation_cost_usd)),
);
properties.insert(
"estimated_cache_read_cost_usd".to_string(),
Value::String(format_usd(cost.cache_read_cost_usd)),
);
if let Some(request_id) = &response.request_id {
properties.insert("request_id".to_string(), Value::String(request_id.clone()));
}
tracer.record_analytics(AnalyticsEvent {
namespace: "api".to_string(),
action: "message_usage".to_string(),
properties,
});
}
fn request_attributes(&self, request: &MessageRequest) -> Map<String, Value> {
let mut attributes = Map::new();
attributes.insert("model".to_string(), Value::String(request.model.clone()));
attributes.insert("stream".to_string(), Value::Bool(request.stream));
attributes.insert("max_tokens".to_string(), Value::from(request.max_tokens));
attributes.insert(
"message_count".to_string(),
Value::from(u64::try_from(request.messages.len()).unwrap_or(u64::MAX)),
);
attributes.insert(
"tool_count".to_string(),
Value::from(
u64::try_from(request.tools.as_ref().map_or(0, Vec::len)).unwrap_or(u64::MAX),
),
);
attributes.insert(
"beta_count".to_string(),
Value::from(u64::try_from(self.request_profile.betas.len()).unwrap_or(u64::MAX)),
);
if !self.request_profile.betas.is_empty() {
attributes.insert(
"betas".to_string(),
Value::Array(
self.request_profile
.betas
.iter()
.cloned()
.map(Value::String)
.collect(),
),
);
}
if !self.request_profile.extra_body.is_empty() {
attributes.insert(
"extra_body_keys".to_string(),
Value::Array(
self.request_profile
.extra_body
.keys()
.cloned()
.map(Value::String)
.collect(),
),
);
}
attributes
}
fn error_attributes(&self, request: &MessageRequest, error: &ApiError) -> Map<String, Value> {
let mut attributes = self.request_attributes(request);
match error {
ApiError::Api {
status,
error_type,
message,
..
} => {
attributes.insert("status".to_string(), Value::from(status.as_u16()));
if let Some(error_type) = error_type {
attributes.insert("error_type".to_string(), Value::String(error_type.clone()));
}
if let Some(message) = message {
attributes.insert("api_message".to_string(), Value::String(message.clone()));
}
}
ApiError::Http(_) => {
attributes.insert("error_type".to_string(), Value::String("http".to_string()));
}
ApiError::Json(_) => {
attributes.insert("error_type".to_string(), Value::String("json".to_string()));
}
_ => {
attributes.insert(
"error_type".to_string(),
Value::String("client".to_string()),
);
}
}
attributes
}
fn backoff_for_attempt(&self, attempt: u32) -> Result<Duration, ApiError> { fn backoff_for_attempt(&self, attempt: u32) -> Result<Duration, ApiError> {
let Some(multiplier) = 1_u32.checked_shl(attempt.saturating_sub(1)) else { let Some(multiplier) = 1_u32.checked_shl(attempt.saturating_sub(1)) else {
return Err(ApiError::BackoffOverflow { return Err(ApiError::BackoffOverflow {

View File

@@ -4,8 +4,8 @@ mod sse;
mod types; mod types;
pub use client::{ pub use client::{
oauth_token_is_expired, read_base_url, resolve_saved_oauth_token, oauth_token_is_expired, read_base_url, resolve_saved_oauth_token, resolve_startup_auth_source,
resolve_startup_auth_source, AnthropicClient, AuthSource, MessageStream, OAuthTokenSet, AnthropicClient, AuthSource, MessageStream, OAuthTokenSet,
}; };
pub use error::ApiError; pub use error::ApiError;
pub use sse::{parse_frame, SseParser}; pub use sse::{parse_frame, SseParser};
@@ -15,3 +15,9 @@ pub use types::{
MessageResponse, MessageStartEvent, MessageStopEvent, OutputContentBlock, StreamEvent, MessageResponse, MessageStartEvent, MessageStopEvent, OutputContentBlock, StreamEvent,
ToolChoice, ToolDefinition, ToolResultContentBlock, Usage, ToolChoice, ToolDefinition, ToolResultContentBlock, Usage,
}; };
pub use telemetry::{
AnalyticsEvent, AnthropicRequestProfile, ClientIdentity, JsonlTelemetrySink,
MemoryTelemetrySink, SessionTraceRecord, SessionTracer, TelemetryEvent, TelemetrySink,
DEFAULT_ANTHROPIC_VERSION,
};

View File

@@ -1,3 +1,4 @@
use runtime::{pricing_for_model, TokenUsage, UsageCostEstimate};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::Value; use serde_json::Value;
@@ -150,7 +151,29 @@ pub struct Usage {
impl Usage { impl Usage {
#[must_use] #[must_use]
pub const fn total_tokens(&self) -> u32 { pub const fn total_tokens(&self) -> u32 {
self.input_tokens + self.output_tokens self.input_tokens
+ self.output_tokens
+ self.cache_creation_input_tokens
+ self.cache_read_input_tokens
}
#[must_use]
pub const fn token_usage(&self) -> TokenUsage {
TokenUsage {
input_tokens: self.input_tokens,
output_tokens: self.output_tokens,
cache_creation_input_tokens: self.cache_creation_input_tokens,
cache_read_input_tokens: self.cache_read_input_tokens,
}
}
#[must_use]
pub fn estimated_cost_usd(&self, model: &str) -> UsageCostEstimate {
let usage = self.token_usage();
pricing_for_model(model).map_or_else(
|| usage.estimate_cost_usd(),
|pricing| usage.estimate_cost_usd_with_pricing(pricing),
)
} }
} }
@@ -210,3 +233,47 @@ pub enum StreamEvent {
ContentBlockStop(ContentBlockStopEvent), ContentBlockStop(ContentBlockStopEvent),
MessageStop(MessageStopEvent), MessageStop(MessageStopEvent),
} }
#[cfg(test)]
mod tests {
use runtime::format_usd;
use super::{MessageResponse, Usage};
#[test]
fn usage_total_tokens_includes_cache_tokens() {
let usage = Usage {
input_tokens: 10,
cache_creation_input_tokens: 2,
cache_read_input_tokens: 3,
output_tokens: 4,
};
assert_eq!(usage.total_tokens(), 19);
assert_eq!(usage.token_usage().total_tokens(), 19);
}
#[test]
fn message_response_estimates_cost_from_model_usage() {
let response = MessageResponse {
id: "msg_cost".to_string(),
kind: "message".to_string(),
role: "assistant".to_string(),
content: Vec::new(),
model: "claude-sonnet-4-20250514".to_string(),
stop_reason: Some("end_turn".to_string()),
stop_sequence: None,
usage: Usage {
input_tokens: 1_000_000,
cache_creation_input_tokens: 100_000,
cache_read_input_tokens: 200_000,
output_tokens: 500_000,
},
request_id: None,
};
let cost = response.usage.estimated_cost_usd(&response.model);
assert_eq!(format_usd(cost.total_cost_usd()), "$54.6750");
assert_eq!(response.total_tokens(), 1_800_000);
}
}

View File

@@ -8,6 +8,7 @@ use api::{
StreamEvent, ToolChoice, ToolDefinition, StreamEvent, ToolChoice, ToolDefinition,
}; };
use serde_json::json; use serde_json::json;
use telemetry::{ClientIdentity, MemoryTelemetrySink, SessionTracer, TelemetryEvent};
use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener; use tokio::net::TcpListener;
use tokio::sync::Mutex; use tokio::sync::Mutex;
@@ -64,6 +65,18 @@ async fn send_message_posts_json_and_parses_response() {
request.headers.get("authorization").map(String::as_str), request.headers.get("authorization").map(String::as_str),
Some("Bearer proxy-token") Some("Bearer proxy-token")
); );
assert_eq!(
request.headers.get("anthropic-version").map(String::as_str),
Some("2023-06-01")
);
assert_eq!(
request.headers.get("user-agent").map(String::as_str),
Some("claude-code/0.1.0")
);
assert_eq!(
request.headers.get("anthropic-beta").map(String::as_str),
Some("claude-code-20250219,prompt-caching-scope-2026-01-05")
);
let body: serde_json::Value = let body: serde_json::Value =
serde_json::from_str(&request.body).expect("request body should be json"); serde_json::from_str(&request.body).expect("request body should be json");
assert_eq!( assert_eq!(
@@ -73,6 +86,115 @@ async fn send_message_posts_json_and_parses_response() {
assert!(body.get("stream").is_none()); assert!(body.get("stream").is_none());
assert_eq!(body["tools"][0]["name"], json!("get_weather")); assert_eq!(body["tools"][0]["name"], json!("get_weather"));
assert_eq!(body["tool_choice"]["type"], json!("auto")); assert_eq!(body["tool_choice"]["type"], json!("auto"));
assert_eq!(
body["betas"],
json!(["claude-code-20250219", "prompt-caching-scope-2026-01-05"])
);
}
#[tokio::test]
async fn send_message_applies_request_profile_and_records_telemetry() {
let state = Arc::new(Mutex::new(Vec::<CapturedRequest>::new()));
let server = spawn_server(
state.clone(),
vec![http_response_with_headers(
"200 OK",
"application/json",
concat!(
"{",
"\"id\":\"msg_profile\",",
"\"type\":\"message\",",
"\"role\":\"assistant\",",
"\"content\":[{\"type\":\"text\",\"text\":\"ok\"}],",
"\"model\":\"claude-3-7-sonnet-latest\",",
"\"stop_reason\":\"end_turn\",",
"\"stop_sequence\":null,",
"\"usage\":{\"input_tokens\":1,\"cache_creation_input_tokens\":2,\"cache_read_input_tokens\":3,\"output_tokens\":1}",
"}"
),
&[("request-id", "req_profile_123")],
)],
)
.await;
let sink = Arc::new(MemoryTelemetrySink::default());
let client = AnthropicClient::new("test-key")
.with_base_url(server.base_url())
.with_client_identity(ClientIdentity::new("claude-code", "9.9.9").with_runtime("rust-cli"))
.with_beta("tools-2026-04-01")
.with_extra_body_param("metadata", json!({"source": "clawd-code"}))
.with_session_tracer(SessionTracer::new("session-telemetry", sink.clone()));
let response = client
.send_message(&sample_request(false))
.await
.expect("request should succeed");
assert_eq!(response.request_id.as_deref(), Some("req_profile_123"));
let captured = state.lock().await;
let request = captured.first().expect("server should capture request");
assert_eq!(
request.headers.get("anthropic-beta").map(String::as_str),
Some("claude-code-20250219,prompt-caching-scope-2026-01-05,tools-2026-04-01")
);
assert_eq!(
request.headers.get("user-agent").map(String::as_str),
Some("claude-code/9.9.9")
);
let body: serde_json::Value =
serde_json::from_str(&request.body).expect("request body should be json");
assert_eq!(body["metadata"]["source"], json!("clawd-code"));
assert_eq!(
body["betas"],
json!([
"claude-code-20250219",
"prompt-caching-scope-2026-01-05",
"tools-2026-04-01"
])
);
let events = sink.events();
assert_eq!(events.len(), 6);
assert!(matches!(
&events[0],
TelemetryEvent::HttpRequestStarted {
session_id,
attempt: 1,
method,
path,
..
} if session_id == "session-telemetry" && method == "POST" && path == "/v1/messages"
));
assert!(matches!(
&events[1],
TelemetryEvent::SessionTrace(trace) if trace.name == "http_request_started"
));
assert!(matches!(
&events[2],
TelemetryEvent::HttpRequestSucceeded {
request_id,
status: 200,
..
} if request_id.as_deref() == Some("req_profile_123")
));
assert!(matches!(
&events[3],
TelemetryEvent::SessionTrace(trace) if trace.name == "http_request_succeeded"
));
assert!(matches!(
&events[4],
TelemetryEvent::Analytics(event)
if event.namespace == "api"
&& event.action == "message_usage"
&& event.properties.get("request_id") == Some(&json!("req_profile_123"))
&& event.properties.get("total_tokens") == Some(&json!(7))
&& event.properties.get("estimated_cost_usd") == Some(&json!("$0.0001"))
));
assert!(matches!(
&events[5],
TelemetryEvent::SessionTrace(trace) if trace.name == "analytics"
));
} }
#[tokio::test] #[tokio::test]

View File

@@ -11,6 +11,7 @@ glob = "0.3"
regex = "1" regex = "1"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
telemetry = { path = "../telemetry" }
tokio = { version = "1", features = ["io-util", "macros", "process", "rt", "rt-multi-thread", "time"] } tokio = { version = "1", features = ["io-util", "macros", "process", "rt", "rt-multi-thread", "time"] }
walkdir = "2" walkdir = "2"

View File

@@ -37,6 +37,7 @@ pub struct RuntimeConfig {
#[derive(Debug, Clone, PartialEq, Eq, Default)] #[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RuntimeFeatureConfig { pub struct RuntimeFeatureConfig {
hooks: RuntimeHookConfig,
mcp: McpConfigCollection, mcp: McpConfigCollection,
oauth: Option<OAuthConfig>, oauth: Option<OAuthConfig>,
model: Option<String>, model: Option<String>,
@@ -44,6 +45,12 @@ pub struct RuntimeFeatureConfig {
sandbox: SandboxConfig, sandbox: SandboxConfig,
} }
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RuntimeHookConfig {
pre_tool_use: Vec<String>,
post_tool_use: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)] #[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct McpConfigCollection { pub struct McpConfigCollection {
servers: BTreeMap<String, ScopedMcpServerConfig>, servers: BTreeMap<String, ScopedMcpServerConfig>,
@@ -221,6 +228,7 @@ impl ConfigLoader {
let merged_value = JsonValue::Object(merged.clone()); let merged_value = JsonValue::Object(merged.clone());
let feature_config = RuntimeFeatureConfig { let feature_config = RuntimeFeatureConfig {
hooks: parse_optional_hooks_config(&merged_value)?,
mcp: McpConfigCollection { mcp: McpConfigCollection {
servers: mcp_servers, servers: mcp_servers,
}, },
@@ -278,6 +286,11 @@ impl RuntimeConfig {
&self.feature_config.mcp &self.feature_config.mcp
} }
#[must_use]
pub fn hooks(&self) -> &RuntimeHookConfig {
&self.feature_config.hooks
}
#[must_use] #[must_use]
pub fn oauth(&self) -> Option<&OAuthConfig> { pub fn oauth(&self) -> Option<&OAuthConfig> {
self.feature_config.oauth.as_ref() self.feature_config.oauth.as_ref()
@@ -300,6 +313,17 @@ impl RuntimeConfig {
} }
impl RuntimeFeatureConfig { impl RuntimeFeatureConfig {
#[must_use]
pub fn with_hooks(mut self, hooks: RuntimeHookConfig) -> Self {
self.hooks = hooks;
self
}
#[must_use]
pub fn hooks(&self) -> &RuntimeHookConfig {
&self.hooks
}
#[must_use] #[must_use]
pub fn mcp(&self) -> &McpConfigCollection { pub fn mcp(&self) -> &McpConfigCollection {
&self.mcp &self.mcp
@@ -326,6 +350,26 @@ impl RuntimeFeatureConfig {
} }
} }
impl RuntimeHookConfig {
#[must_use]
pub fn new(pre_tool_use: Vec<String>, post_tool_use: Vec<String>) -> Self {
Self {
pre_tool_use,
post_tool_use,
}
}
#[must_use]
pub fn pre_tool_use(&self) -> &[String] {
&self.pre_tool_use
}
#[must_use]
pub fn post_tool_use(&self) -> &[String] {
&self.post_tool_use
}
}
impl McpConfigCollection { impl McpConfigCollection {
#[must_use] #[must_use]
pub fn servers(&self) -> &BTreeMap<String, ScopedMcpServerConfig> { pub fn servers(&self) -> &BTreeMap<String, ScopedMcpServerConfig> {
@@ -424,6 +468,22 @@ fn parse_optional_model(root: &JsonValue) -> Option<String> {
.map(ToOwned::to_owned) .map(ToOwned::to_owned)
} }
fn parse_optional_hooks_config(root: &JsonValue) -> Result<RuntimeHookConfig, ConfigError> {
let Some(object) = root.as_object() else {
return Ok(RuntimeHookConfig::default());
};
let Some(hooks_value) = object.get("hooks") else {
return Ok(RuntimeHookConfig::default());
};
let hooks = expect_object(hooks_value, "merged settings.hooks")?;
Ok(RuntimeHookConfig {
pre_tool_use: optional_string_array(hooks, "PreToolUse", "merged settings.hooks")?
.unwrap_or_default(),
post_tool_use: optional_string_array(hooks, "PostToolUse", "merged settings.hooks")?
.unwrap_or_default(),
})
}
fn parse_optional_permission_mode( fn parse_optional_permission_mode(
root: &JsonValue, root: &JsonValue,
) -> Result<Option<ResolvedPermissionMode>, ConfigError> { ) -> Result<Option<ResolvedPermissionMode>, ConfigError> {
@@ -836,6 +896,8 @@ mod tests {
.and_then(JsonValue::as_object) .and_then(JsonValue::as_object)
.expect("hooks object") .expect("hooks object")
.contains_key("PostToolUse")); .contains_key("PostToolUse"));
assert_eq!(loaded.hooks().pre_tool_use(), &["base".to_string()]);
assert_eq!(loaded.hooks().post_tool_use(), &["project".to_string()]);
assert!(loaded.mcp().get("home").is_some()); assert!(loaded.mcp().get("home").is_some());
assert!(loaded.mcp().get("project").is_some()); assert!(loaded.mcp().get("project").is_some());

View File

@@ -1,9 +1,14 @@
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::fmt::{Display, Formatter}; use std::fmt::{Display, Formatter};
use serde_json::{Map, Value};
use telemetry::SessionTracer;
use crate::compact::{ use crate::compact::{
compact_session, estimate_session_tokens, CompactionConfig, CompactionResult, compact_session, estimate_session_tokens, CompactionConfig, CompactionResult,
}; };
use crate::config::RuntimeFeatureConfig;
use crate::hooks::{HookRunResult, HookRunner};
use crate::permissions::{PermissionOutcome, PermissionPolicy, PermissionPrompter}; use crate::permissions::{PermissionOutcome, PermissionPolicy, PermissionPrompter};
use crate::session::{ContentBlock, ConversationMessage, Session}; use crate::session::{ContentBlock, ConversationMessage, Session};
use crate::usage::{TokenUsage, UsageTracker}; use crate::usage::{TokenUsage, UsageTracker};
@@ -94,6 +99,8 @@ pub struct ConversationRuntime<C, T> {
system_prompt: Vec<String>, system_prompt: Vec<String>,
max_iterations: usize, max_iterations: usize,
usage_tracker: UsageTracker, usage_tracker: UsageTracker,
hook_runner: HookRunner,
session_tracer: Option<SessionTracer>,
} }
impl<C, T> ConversationRuntime<C, T> impl<C, T> ConversationRuntime<C, T>
@@ -108,6 +115,25 @@ where
tool_executor: T, tool_executor: T,
permission_policy: PermissionPolicy, permission_policy: PermissionPolicy,
system_prompt: Vec<String>, system_prompt: Vec<String>,
) -> Self {
Self::new_with_features(
session,
api_client,
tool_executor,
permission_policy,
system_prompt,
&RuntimeFeatureConfig::default(),
)
}
#[must_use]
pub fn new_with_features(
session: Session,
api_client: C,
tool_executor: T,
permission_policy: PermissionPolicy,
system_prompt: Vec<String>,
feature_config: &RuntimeFeatureConfig,
) -> Self { ) -> Self {
let usage_tracker = UsageTracker::from_session(&session); let usage_tracker = UsageTracker::from_session(&session);
Self { Self {
@@ -118,6 +144,8 @@ where
system_prompt, system_prompt,
max_iterations: usize::MAX, max_iterations: usize::MAX,
usage_tracker, usage_tracker,
hook_runner: HookRunner::from_feature_config(feature_config),
session_tracer: None,
} }
} }
@@ -127,14 +155,23 @@ where
self self
} }
#[must_use]
pub fn with_session_tracer(mut self, session_tracer: SessionTracer) -> Self {
self.session_tracer = Some(session_tracer);
self
}
#[allow(clippy::too_many_lines)]
pub fn run_turn( pub fn run_turn(
&mut self, &mut self,
user_input: impl Into<String>, user_input: impl Into<String>,
mut prompter: Option<&mut dyn PermissionPrompter>, mut prompter: Option<&mut dyn PermissionPrompter>,
) -> Result<TurnSummary, RuntimeError> { ) -> Result<TurnSummary, RuntimeError> {
let user_input = user_input.into();
self.record_turn_started(&user_input);
self.session self.session
.messages .messages
.push(ConversationMessage::user_text(user_input.into())); .push(ConversationMessage::user_text(user_input));
let mut assistant_messages = Vec::new(); let mut assistant_messages = Vec::new();
let mut tool_results = Vec::new(); let mut tool_results = Vec::new();
@@ -143,17 +180,31 @@ where
loop { loop {
iterations += 1; iterations += 1;
if iterations > self.max_iterations { if iterations > self.max_iterations {
return Err(RuntimeError::new( let error = RuntimeError::new(
"conversation loop exceeded the maximum number of iterations", "conversation loop exceeded the maximum number of iterations",
)); );
self.record_turn_failed(iterations, &error);
return Err(error);
} }
let request = ApiRequest { let request = ApiRequest {
system_prompt: self.system_prompt.clone(), system_prompt: self.system_prompt.clone(),
messages: self.session.messages.clone(), messages: self.session.messages.clone(),
}; };
let events = self.api_client.stream(request)?; let events = match self.api_client.stream(request) {
let (assistant_message, usage) = build_assistant_message(events)?; Ok(events) => events,
Err(error) => {
self.record_turn_failed(iterations, &error);
return Err(error);
}
};
let (assistant_message, usage) = match build_assistant_message(events) {
Ok(result) => result,
Err(error) => {
self.record_turn_failed(iterations, &error);
return Err(error);
}
};
if let Some(usage) = usage { if let Some(usage) = usage {
self.usage_tracker.record(usage); self.usage_tracker.record(usage);
} }
@@ -167,6 +218,11 @@ where
_ => None, _ => None,
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();
self.record_assistant_iteration(
iterations,
&assistant_message,
pending_tool_uses.len(),
);
self.session.messages.push(assistant_message.clone()); self.session.messages.push(assistant_message.clone());
assistant_messages.push(assistant_message); assistant_messages.push(assistant_message);
@@ -176,6 +232,7 @@ where
} }
for (tool_use_id, tool_name, input) in pending_tool_uses { for (tool_use_id, tool_name, input) in pending_tool_uses {
self.record_tool_started(iterations, &tool_name);
let permission_outcome = if let Some(prompt) = prompter.as_mut() { let permission_outcome = if let Some(prompt) = prompter.as_mut() {
self.permission_policy self.permission_policy
.authorize(&tool_name, &input, Some(*prompt)) .authorize(&tool_name, &input, Some(*prompt))
@@ -185,36 +242,61 @@ where
let result_message = match permission_outcome { let result_message = match permission_outcome {
PermissionOutcome::Allow => { PermissionOutcome::Allow => {
let pre_hook_result = self.hook_runner.run_pre_tool_use(&tool_name, &input);
if pre_hook_result.is_denied() {
let deny_message = format!("PreToolUse hook denied tool `{tool_name}`");
ConversationMessage::tool_result(
tool_use_id,
tool_name,
format_hook_message(&pre_hook_result, &deny_message),
true,
)
} else {
let (mut output, mut is_error) =
match self.tool_executor.execute(&tool_name, &input) { match self.tool_executor.execute(&tool_name, &input) {
Ok(output) => ConversationMessage::tool_result( Ok(output) => (output, false),
Err(error) => (error.to_string(), true),
};
output = merge_hook_feedback(pre_hook_result.messages(), output, false);
let post_hook_result = self
.hook_runner
.run_post_tool_use(&tool_name, &input, &output, is_error);
if post_hook_result.is_denied() {
is_error = true;
}
output = merge_hook_feedback(
post_hook_result.messages(),
output,
post_hook_result.is_denied(),
);
ConversationMessage::tool_result(
tool_use_id, tool_use_id,
tool_name, tool_name,
output, output,
false, is_error,
), )
Err(error) => ConversationMessage::tool_result(
tool_use_id,
tool_name,
error.to_string(),
true,
),
} }
} }
PermissionOutcome::Deny { reason } => { PermissionOutcome::Deny { reason } => {
ConversationMessage::tool_result(tool_use_id, tool_name, reason, true) ConversationMessage::tool_result(tool_use_id, tool_name, reason, true)
} }
}; };
self.record_tool_finished(iterations, &result_message);
self.session.messages.push(result_message.clone()); self.session.messages.push(result_message.clone());
tool_results.push(result_message); tool_results.push(result_message);
} }
} }
Ok(TurnSummary { let summary = TurnSummary {
assistant_messages, assistant_messages,
tool_results, tool_results,
iterations, iterations,
usage: self.usage_tracker.cumulative_usage(), usage: self.usage_tracker.cumulative_usage(),
}) };
self.record_turn_completed(&summary);
Ok(summary)
} }
#[must_use] #[must_use]
@@ -241,6 +323,126 @@ where
pub fn into_session(self) -> Session { pub fn into_session(self) -> Session {
self.session self.session
} }
fn record_turn_started(&self, user_input: &str) {
if let Some(tracer) = &self.session_tracer {
let mut attributes = Map::new();
attributes.insert(
"message_count_before".to_string(),
Value::from(u64::try_from(self.session.messages.len()).unwrap_or(u64::MAX)),
);
attributes.insert(
"input_chars".to_string(),
Value::from(u64::try_from(user_input.chars().count()).unwrap_or(u64::MAX)),
);
tracer.record("turn_started", attributes);
}
}
fn record_assistant_iteration(
&self,
iteration: usize,
assistant_message: &ConversationMessage,
pending_tool_count: usize,
) {
if let Some(tracer) = &self.session_tracer {
let mut attributes = Map::new();
attributes.insert(
"iteration".to_string(),
Value::from(u64::try_from(iteration).unwrap_or(u64::MAX)),
);
attributes.insert(
"block_count".to_string(),
Value::from(u64::try_from(assistant_message.blocks.len()).unwrap_or(u64::MAX)),
);
attributes.insert(
"pending_tool_count".to_string(),
Value::from(u64::try_from(pending_tool_count).unwrap_or(u64::MAX)),
);
tracer.record("assistant_iteration_completed", attributes);
}
}
fn record_tool_started(&self, iteration: usize, tool_name: &str) {
if let Some(tracer) = &self.session_tracer {
let mut attributes = Map::new();
attributes.insert(
"iteration".to_string(),
Value::from(u64::try_from(iteration).unwrap_or(u64::MAX)),
);
attributes.insert(
"tool_name".to_string(),
Value::String(tool_name.to_string()),
);
tracer.record("tool_execution_started", attributes);
}
}
fn record_tool_finished(&self, iteration: usize, result_message: &ConversationMessage) {
let Some(tracer) = &self.session_tracer else {
return;
};
let Some(ContentBlock::ToolResult {
tool_name,
is_error,
output,
..
}) = result_message.blocks.first()
else {
return;
};
let mut attributes = Map::new();
attributes.insert(
"iteration".to_string(),
Value::from(u64::try_from(iteration).unwrap_or(u64::MAX)),
);
attributes.insert("tool_name".to_string(), Value::String(tool_name.clone()));
attributes.insert("is_error".to_string(), Value::Bool(*is_error));
attributes.insert(
"output_chars".to_string(),
Value::from(u64::try_from(output.chars().count()).unwrap_or(u64::MAX)),
);
tracer.record("tool_execution_finished", attributes);
}
fn record_turn_completed(&self, summary: &TurnSummary) {
if let Some(tracer) = &self.session_tracer {
let mut attributes = Map::new();
attributes.insert(
"assistant_message_count".to_string(),
Value::from(u64::try_from(summary.assistant_messages.len()).unwrap_or(u64::MAX)),
);
attributes.insert(
"tool_result_count".to_string(),
Value::from(u64::try_from(summary.tool_results.len()).unwrap_or(u64::MAX)),
);
attributes.insert(
"iterations".to_string(),
Value::from(u64::try_from(summary.iterations).unwrap_or(u64::MAX)),
);
attributes.insert(
"total_input_tokens".to_string(),
Value::from(summary.usage.input_tokens),
);
attributes.insert(
"total_output_tokens".to_string(),
Value::from(summary.usage.output_tokens),
);
tracer.record("turn_completed", attributes);
}
}
fn record_turn_failed(&self, iteration: usize, error: &RuntimeError) {
if let Some(tracer) = &self.session_tracer {
let mut attributes = Map::new();
attributes.insert(
"iteration".to_string(),
Value::from(u64::try_from(iteration).unwrap_or(u64::MAX)),
);
attributes.insert("error".to_string(), Value::String(error.to_string()));
tracer.record("turn_failed", attributes);
}
}
} }
fn build_assistant_message( fn build_assistant_message(
@@ -290,6 +492,32 @@ fn flush_text_block(text: &mut String, blocks: &mut Vec<ContentBlock>) {
} }
} }
fn format_hook_message(result: &HookRunResult, fallback: &str) -> String {
if result.messages().is_empty() {
fallback.to_string()
} else {
result.messages().join("\n")
}
}
fn merge_hook_feedback(messages: &[String], output: String, denied: bool) -> String {
if messages.is_empty() {
return output;
}
let mut sections = Vec::new();
if !output.trim().is_empty() {
sections.push(output);
}
let label = if denied {
"Hook feedback (denied)"
} else {
"Hook feedback"
};
sections.push(format!("{label}:\n{}", messages.join("\n")));
sections.join("\n\n")
}
type ToolHandler = Box<dyn FnMut(&str) -> Result<String, ToolError>>; type ToolHandler = Box<dyn FnMut(&str) -> Result<String, ToolError>>;
#[derive(Default)] #[derive(Default)]
@@ -329,6 +557,7 @@ mod tests {
StaticToolExecutor, StaticToolExecutor,
}; };
use crate::compact::CompactionConfig; use crate::compact::CompactionConfig;
use crate::config::{RuntimeFeatureConfig, RuntimeHookConfig};
use crate::permissions::{ use crate::permissions::{
PermissionMode, PermissionPolicy, PermissionPromptDecision, PermissionPrompter, PermissionMode, PermissionPolicy, PermissionPromptDecision, PermissionPrompter,
PermissionRequest, PermissionRequest,
@@ -337,6 +566,8 @@ mod tests {
use crate::session::{ContentBlock, MessageRole, Session}; use crate::session::{ContentBlock, MessageRole, Session};
use crate::usage::TokenUsage; use crate::usage::TokenUsage;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc;
use telemetry::{MemoryTelemetrySink, SessionTracer, TelemetryEvent};
struct ScriptedApiClient { struct ScriptedApiClient {
call_count: usize, call_count: usize,
@@ -449,6 +680,39 @@ mod tests {
)); ));
} }
#[test]
fn records_runtime_session_trace_events() {
let sink = Arc::new(MemoryTelemetrySink::default());
let tracer = SessionTracer::new("session-runtime", sink.clone());
let mut runtime = ConversationRuntime::new(
Session::new(),
ScriptedApiClient { call_count: 0 },
StaticToolExecutor::new().register("add", |_input| Ok("4".to_string())),
PermissionPolicy::new(PermissionMode::WorkspaceWrite),
vec!["system".to_string()],
)
.with_session_tracer(tracer);
runtime
.run_turn("what is 2 + 2?", Some(&mut PromptAllowOnce))
.expect("conversation loop should succeed");
let events = sink.events();
let trace_names = events
.iter()
.filter_map(|event| match event {
TelemetryEvent::SessionTrace(trace) => Some(trace.name.as_str()),
_ => None,
})
.collect::<Vec<_>>();
assert!(trace_names.contains(&"turn_started"));
assert!(trace_names.contains(&"assistant_iteration_completed"));
assert!(trace_names.contains(&"tool_execution_started"));
assert!(trace_names.contains(&"tool_execution_finished"));
assert!(trace_names.contains(&"turn_completed"));
}
#[test] #[test]
fn records_denied_tool_results_when_prompt_rejects() { fn records_denied_tool_results_when_prompt_rejects() {
struct RejectPrompter; struct RejectPrompter;
@@ -503,6 +767,141 @@ mod tests {
)); ));
} }
#[test]
fn denies_tool_use_when_pre_tool_hook_blocks() {
struct SingleCallApiClient;
impl ApiClient for SingleCallApiClient {
fn stream(&mut self, request: ApiRequest) -> Result<Vec<AssistantEvent>, RuntimeError> {
if request
.messages
.iter()
.any(|message| message.role == MessageRole::Tool)
{
return Ok(vec![
AssistantEvent::TextDelta("blocked".to_string()),
AssistantEvent::MessageStop,
]);
}
Ok(vec![
AssistantEvent::ToolUse {
id: "tool-1".to_string(),
name: "blocked".to_string(),
input: r#"{"path":"secret.txt"}"#.to_string(),
},
AssistantEvent::MessageStop,
])
}
}
let mut runtime = ConversationRuntime::new_with_features(
Session::new(),
SingleCallApiClient,
StaticToolExecutor::new().register("blocked", |_input| {
panic!("tool should not execute when hook denies")
}),
PermissionPolicy::new(PermissionMode::DangerFullAccess),
vec!["system".to_string()],
&RuntimeFeatureConfig::default().with_hooks(RuntimeHookConfig::new(
vec![shell_snippet("printf 'blocked by hook'; exit 2")],
Vec::new(),
)),
);
let summary = runtime
.run_turn("use the tool", None)
.expect("conversation should continue after hook denial");
assert_eq!(summary.tool_results.len(), 1);
let ContentBlock::ToolResult {
is_error, output, ..
} = &summary.tool_results[0].blocks[0]
else {
panic!("expected tool result block");
};
assert!(
*is_error,
"hook denial should produce an error result: {output}"
);
assert!(
output.contains("denied tool") || output.contains("blocked by hook"),
"unexpected hook denial output: {output:?}"
);
}
#[test]
fn appends_post_tool_hook_feedback_to_tool_result() {
struct TwoCallApiClient {
calls: usize,
}
impl ApiClient for TwoCallApiClient {
fn stream(&mut self, request: ApiRequest) -> Result<Vec<AssistantEvent>, RuntimeError> {
self.calls += 1;
match self.calls {
1 => Ok(vec![
AssistantEvent::ToolUse {
id: "tool-1".to_string(),
name: "add".to_string(),
input: r#"{"lhs":2,"rhs":2}"#.to_string(),
},
AssistantEvent::MessageStop,
]),
2 => {
assert!(request
.messages
.iter()
.any(|message| message.role == MessageRole::Tool));
Ok(vec![
AssistantEvent::TextDelta("done".to_string()),
AssistantEvent::MessageStop,
])
}
_ => Err(RuntimeError::new("unexpected extra API call")),
}
}
}
let mut runtime = ConversationRuntime::new_with_features(
Session::new(),
TwoCallApiClient { calls: 0 },
StaticToolExecutor::new().register("add", |_input| Ok("4".to_string())),
PermissionPolicy::new(PermissionMode::DangerFullAccess),
vec!["system".to_string()],
&RuntimeFeatureConfig::default().with_hooks(RuntimeHookConfig::new(
vec![shell_snippet("printf 'pre hook ran'")],
vec![shell_snippet("printf 'post hook ran'")],
)),
);
let summary = runtime
.run_turn("use add", None)
.expect("tool loop succeeds");
assert_eq!(summary.tool_results.len(), 1);
let ContentBlock::ToolResult {
is_error, output, ..
} = &summary.tool_results[0].blocks[0]
else {
panic!("expected tool result block");
};
assert!(
!*is_error,
"post hook should preserve non-error result: {output:?}"
);
assert!(
output.contains('4'),
"tool output missing value: {output:?}"
);
assert!(
output.contains("pre hook ran"),
"tool output missing pre hook feedback: {output:?}"
);
assert!(
output.contains("post hook ran"),
"tool output missing post hook feedback: {output:?}"
);
}
#[test] #[test]
fn reconstructs_usage_tracker_from_restored_session() { fn reconstructs_usage_tracker_from_restored_session() {
struct SimpleApi; struct SimpleApi;
@@ -581,4 +980,14 @@ mod tests {
MessageRole::System MessageRole::System
); );
} }
#[cfg(windows)]
fn shell_snippet(script: &str) -> String {
script.replace('\'', "\"")
}
#[cfg(not(windows))]
fn shell_snippet(script: &str) -> String {
script.to_string()
}
} }

View File

@@ -0,0 +1,353 @@
use std::ffi::OsStr;
use std::process::Command;
use serde_json::json;
use crate::config::{RuntimeFeatureConfig, RuntimeHookConfig};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HookEvent {
PreToolUse,
PostToolUse,
}
impl HookEvent {
fn as_str(self) -> &'static str {
match self {
Self::PreToolUse => "PreToolUse",
Self::PostToolUse => "PostToolUse",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HookRunResult {
denied: bool,
messages: Vec<String>,
}
impl HookRunResult {
#[must_use]
pub fn allow(messages: Vec<String>) -> Self {
Self {
denied: false,
messages,
}
}
#[must_use]
pub fn is_denied(&self) -> bool {
self.denied
}
#[must_use]
pub fn messages(&self) -> &[String] {
&self.messages
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct HookRunner {
config: RuntimeHookConfig,
}
impl HookRunner {
#[must_use]
pub fn new(config: RuntimeHookConfig) -> Self {
Self { config }
}
#[must_use]
pub fn from_feature_config(feature_config: &RuntimeFeatureConfig) -> Self {
Self::new(feature_config.hooks().clone())
}
#[must_use]
pub fn run_pre_tool_use(&self, tool_name: &str, tool_input: &str) -> HookRunResult {
Self::run_commands(
HookEvent::PreToolUse,
self.config.pre_tool_use(),
tool_name,
tool_input,
None,
false,
)
}
#[must_use]
pub fn run_post_tool_use(
&self,
tool_name: &str,
tool_input: &str,
tool_output: &str,
is_error: bool,
) -> HookRunResult {
Self::run_commands(
HookEvent::PostToolUse,
self.config.post_tool_use(),
tool_name,
tool_input,
Some(tool_output),
is_error,
)
}
fn run_commands(
event: HookEvent,
commands: &[String],
tool_name: &str,
tool_input: &str,
tool_output: Option<&str>,
is_error: bool,
) -> HookRunResult {
if commands.is_empty() {
return HookRunResult::allow(Vec::new());
}
let payload = json!({
"hook_event_name": event.as_str(),
"tool_name": tool_name,
"tool_input": parse_tool_input(tool_input),
"tool_input_json": tool_input,
"tool_output": tool_output,
"tool_result_is_error": is_error,
})
.to_string();
let invocation = HookInvocation {
event,
tool_name,
tool_input,
tool_output,
is_error,
payload: &payload,
};
let mut messages = Vec::new();
for command in commands {
match Self::run_command(command, &invocation) {
HookCommandOutcome::Allow { message } => {
if let Some(message) = message {
messages.push(message);
}
}
HookCommandOutcome::Deny { message } => {
let message = message.unwrap_or_else(|| {
format!("{} hook denied tool `{tool_name}`", event.as_str())
});
messages.push(message);
return HookRunResult {
denied: true,
messages,
};
}
HookCommandOutcome::Warn { message } => messages.push(message),
}
}
HookRunResult::allow(messages)
}
fn run_command(command: &str, invocation: &HookInvocation<'_>) -> HookCommandOutcome {
let mut child = shell_command(command);
child.stdin(std::process::Stdio::piped());
child.stdout(std::process::Stdio::piped());
child.stderr(std::process::Stdio::piped());
child.env("HOOK_EVENT", invocation.event.as_str());
child.env("HOOK_TOOL_NAME", invocation.tool_name);
child.env("HOOK_TOOL_INPUT", invocation.tool_input);
child.env(
"HOOK_TOOL_IS_ERROR",
if invocation.is_error { "1" } else { "0" },
);
if let Some(tool_output) = invocation.tool_output {
child.env("HOOK_TOOL_OUTPUT", tool_output);
}
match child.output_with_stdin(invocation.payload.as_bytes()) {
Ok(output) => {
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let message = (!stdout.is_empty()).then_some(stdout);
match output.status.code() {
Some(0) => HookCommandOutcome::Allow { message },
Some(2) => HookCommandOutcome::Deny { message },
Some(code) => HookCommandOutcome::Warn {
message: format_hook_warning(
command,
code,
message.as_deref(),
stderr.as_str(),
),
},
None => HookCommandOutcome::Warn {
message: format!(
"{} hook `{command}` terminated by signal while handling `{}`",
invocation.event.as_str(),
invocation.tool_name
),
},
}
}
Err(error) => HookCommandOutcome::Warn {
message: format!(
"{} hook `{command}` failed to start for `{tool_name}`: {error}",
invocation.event.as_str(),
tool_name = invocation.tool_name
),
},
}
}
}
struct HookInvocation<'a> {
event: HookEvent,
tool_name: &'a str,
tool_input: &'a str,
tool_output: Option<&'a str>,
is_error: bool,
payload: &'a str,
}
enum HookCommandOutcome {
Allow { message: Option<String> },
Deny { message: Option<String> },
Warn { message: String },
}
fn parse_tool_input(tool_input: &str) -> serde_json::Value {
serde_json::from_str(tool_input).unwrap_or_else(|_| json!({ "raw": tool_input }))
}
fn format_hook_warning(command: &str, code: i32, stdout: Option<&str>, stderr: &str) -> String {
let mut message =
format!("Hook `{command}` exited with status {code}; allowing tool execution to continue");
if let Some(stdout) = stdout.filter(|stdout| !stdout.is_empty()) {
message.push_str(": ");
message.push_str(stdout);
} else if !stderr.is_empty() {
message.push_str(": ");
message.push_str(stderr);
}
message
}
fn shell_command(command: &str) -> CommandWithStdin {
#[cfg(windows)]
let mut command_builder = {
let mut command_builder = Command::new("cmd");
command_builder.arg("/C").arg(command);
CommandWithStdin::new(command_builder)
};
#[cfg(not(windows))]
let command_builder = {
let mut command_builder = Command::new("sh");
command_builder.arg("-lc").arg(command);
CommandWithStdin::new(command_builder)
};
command_builder
}
struct CommandWithStdin {
command: Command,
}
impl CommandWithStdin {
fn new(command: Command) -> Self {
Self { command }
}
fn stdin(&mut self, cfg: std::process::Stdio) -> &mut Self {
self.command.stdin(cfg);
self
}
fn stdout(&mut self, cfg: std::process::Stdio) -> &mut Self {
self.command.stdout(cfg);
self
}
fn stderr(&mut self, cfg: std::process::Stdio) -> &mut Self {
self.command.stderr(cfg);
self
}
fn env<K, V>(&mut self, key: K, value: V) -> &mut Self
where
K: AsRef<OsStr>,
V: AsRef<OsStr>,
{
self.command.env(key, value);
self
}
fn output_with_stdin(&mut self, stdin: &[u8]) -> std::io::Result<std::process::Output> {
let mut child = self.command.spawn()?;
if let Some(mut child_stdin) = child.stdin.take() {
use std::io::Write;
child_stdin.write_all(stdin)?;
}
child.wait_with_output()
}
}
#[cfg(test)]
mod tests {
use super::{HookRunResult, HookRunner};
use crate::config::{RuntimeFeatureConfig, RuntimeHookConfig};
#[test]
fn allows_exit_code_zero_and_captures_stdout() {
let runner = HookRunner::new(RuntimeHookConfig::new(
vec![shell_snippet("printf 'pre ok'")],
Vec::new(),
));
let result = runner.run_pre_tool_use("Read", r#"{"path":"README.md"}"#);
assert_eq!(result, HookRunResult::allow(vec!["pre ok".to_string()]));
}
#[test]
fn denies_exit_code_two() {
let runner = HookRunner::new(RuntimeHookConfig::new(
vec![shell_snippet("printf 'blocked by hook'; exit 2")],
Vec::new(),
));
let result = runner.run_pre_tool_use("Bash", r#"{"command":"pwd"}"#);
assert!(result.is_denied());
assert_eq!(result.messages(), &["blocked by hook".to_string()]);
}
#[test]
fn warns_for_other_non_zero_statuses() {
let runner = HookRunner::from_feature_config(&RuntimeFeatureConfig::default().with_hooks(
RuntimeHookConfig::new(
vec![shell_snippet("printf 'warning hook'; exit 1")],
Vec::new(),
),
));
let result = runner.run_pre_tool_use("Edit", r#"{"file":"src/lib.rs"}"#);
assert!(!result.is_denied());
assert!(result
.messages()
.iter()
.any(|message| message.contains("allowing tool execution to continue")));
}
#[cfg(windows)]
fn shell_snippet(script: &str) -> String {
script.replace('\'', "\"")
}
#[cfg(not(windows))]
fn shell_snippet(script: &str) -> String {
script.to_string()
}
}

View File

@@ -4,6 +4,7 @@ mod compact;
mod config; mod config;
mod conversation; mod conversation;
mod file_ops; mod file_ops;
mod hooks;
mod json; mod json;
mod mcp; mod mcp;
mod mcp_client; mod mcp_client;
@@ -26,8 +27,8 @@ pub use config::{
ConfigEntry, ConfigError, ConfigLoader, ConfigSource, McpClaudeAiProxyServerConfig, ConfigEntry, ConfigError, ConfigLoader, ConfigSource, McpClaudeAiProxyServerConfig,
McpConfigCollection, McpOAuthConfig, McpRemoteServerConfig, McpSdkServerConfig, McpConfigCollection, McpOAuthConfig, McpRemoteServerConfig, McpSdkServerConfig,
McpServerConfig, McpStdioServerConfig, McpTransport, McpWebSocketServerConfig, OAuthConfig, McpServerConfig, McpStdioServerConfig, McpTransport, McpWebSocketServerConfig, OAuthConfig,
ResolvedPermissionMode, RuntimeConfig, RuntimeFeatureConfig, ScopedMcpServerConfig, ResolvedPermissionMode, RuntimeConfig, RuntimeFeatureConfig, RuntimeHookConfig,
CLAUDE_CODE_SETTINGS_SCHEMA_NAME, ScopedMcpServerConfig, CLAUDE_CODE_SETTINGS_SCHEMA_NAME,
}; };
pub use conversation::{ pub use conversation::{
ApiClient, ApiRequest, AssistantEvent, ConversationRuntime, RuntimeError, StaticToolExecutor, ApiClient, ApiRequest, AssistantEvent, ConversationRuntime, RuntimeError, StaticToolExecutor,
@@ -38,6 +39,7 @@ pub use file_ops::{
GrepSearchInput, GrepSearchOutput, ReadFileOutput, StructuredPatchHunk, TextFilePayload, GrepSearchInput, GrepSearchOutput, ReadFileOutput, StructuredPatchHunk, TextFilePayload,
WriteFileOutput, WriteFileOutput,
}; };
pub use hooks::{HookEvent, HookRunResult, HookRunner};
pub use mcp::{ pub use mcp::{
mcp_server_signature, mcp_tool_name, mcp_tool_prefix, normalize_name_for_mcp, mcp_server_signature, mcp_tool_name, mcp_tool_prefix, normalize_name_for_mcp,
scoped_mcp_config_hash, unwrap_ccr_proxy_url, scoped_mcp_config_hash, unwrap_ccr_proxy_url,

View File

@@ -1144,8 +1144,20 @@ mod tests {
} }
fn cleanup_script(script_path: &Path) { fn cleanup_script(script_path: &Path) {
fs::remove_file(script_path).expect("cleanup script"); if let Err(error) = fs::remove_file(script_path) {
fs::remove_dir_all(script_path.parent().expect("script parent")).expect("cleanup dir"); assert_eq!(
error.kind(),
std::io::ErrorKind::NotFound,
"cleanup script: {error}"
);
}
if let Err(error) = fs::remove_dir_all(script_path.parent().expect("script parent")) {
assert_eq!(
error.kind(),
std::io::ErrorKind::NotFound,
"cleanup dir: {error}"
);
}
} }
fn manager_server_config( fn manager_server_config(

View File

@@ -4,6 +4,7 @@ mod render;
use std::collections::{BTreeMap, BTreeSet}; use std::collections::{BTreeMap, BTreeSet};
use std::env; use std::env;
use std::fmt::Write as _;
use std::fs; use std::fs;
use std::io::{self, Read, Write}; use std::io::{self, Read, Write};
use std::net::TcpListener; use std::net::TcpListener;
@@ -13,8 +14,9 @@ use std::time::{SystemTime, UNIX_EPOCH};
use api::{ use api::{
resolve_startup_auth_source, AnthropicClient, AuthSource, ContentBlockDelta, InputContentBlock, resolve_startup_auth_source, AnthropicClient, AuthSource, ContentBlockDelta, InputContentBlock,
InputMessage, MessageRequest, MessageResponse, OutputContentBlock, InputMessage, JsonlTelemetrySink, MessageRequest, MessageResponse, OutputContentBlock,
StreamEvent as ApiStreamEvent, ToolChoice, ToolDefinition, ToolResultContentBlock, SessionTracer, StreamEvent as ApiStreamEvent, ToolChoice, ToolDefinition,
ToolResultContentBlock,
}; };
use commands::{ use commands::{
@@ -27,7 +29,7 @@ use runtime::{
clear_oauth_credentials, generate_pkce_pair, generate_state, load_system_prompt, clear_oauth_credentials, generate_pkce_pair, generate_state, load_system_prompt,
parse_oauth_callback_request_target, save_oauth_credentials, ApiClient, ApiRequest, parse_oauth_callback_request_target, save_oauth_credentials, ApiClient, ApiRequest,
AssistantEvent, CompactionConfig, ConfigLoader, ConfigSource, ContentBlock, AssistantEvent, CompactionConfig, ConfigLoader, ConfigSource, ContentBlock,
ConversationMessage, ConversationRuntime, MessageRole, OAuthAuthorizationRequest, ConversationMessage, ConversationRuntime, MessageRole, OAuthAuthorizationRequest, OAuthConfig,
OAuthTokenExchangeRequest, PermissionMode, PermissionPolicy, ProjectContext, RuntimeError, OAuthTokenExchangeRequest, PermissionMode, PermissionPolicy, ProjectContext, RuntimeError,
Session, TokenUsage, ToolError, ToolExecutor, UsageTracker, Session, TokenUsage, ToolError, ToolExecutor, UsageTracker,
}; };
@@ -44,6 +46,7 @@ fn max_tokens_for_model(model: &str) -> u32 {
} }
const DEFAULT_DATE: &str = "2026-03-31"; const DEFAULT_DATE: &str = "2026-03-31";
const DEFAULT_OAUTH_CALLBACK_PORT: u16 = 4545; const DEFAULT_OAUTH_CALLBACK_PORT: u16 = 4545;
const TELEMETRY_LOG_PATH_ENV: &str = "CLAW_TELEMETRY_LOG_PATH";
const VERSION: &str = env!("CARGO_PKG_VERSION"); const VERSION: &str = env!("CARGO_PKG_VERSION");
const BUILD_TARGET: Option<&str> = option_env!("TARGET"); const BUILD_TARGET: Option<&str> = option_env!("TARGET");
const GIT_SHA: Option<&str> = option_env!("GIT_SHA"); const GIT_SHA: Option<&str> = option_env!("GIT_SHA");
@@ -196,6 +199,25 @@ fn parse_args(args: &[String]) -> Result<CliAction, String> {
permission_mode = PermissionMode::DangerFullAccess; permission_mode = PermissionMode::DangerFullAccess;
index += 1; index += 1;
} }
"-p" => {
// Claude Code compat: -p "prompt" = one-shot prompt
let prompt = args[index + 1..].join(" ");
if prompt.trim().is_empty() {
return Err("-p requires a prompt string".to_string());
}
return Ok(CliAction::Prompt {
prompt,
model: resolve_model_alias(&model).to_string(),
output_format,
allowed_tools: normalize_allowed_tools(&allowed_tool_values)?,
permission_mode,
});
}
"--print" => {
// Claude Code compat: --print makes output non-interactive
output_format = CliOutputFormat::Text;
index += 1;
}
"--allowedTools" | "--allowed-tools" => { "--allowedTools" | "--allowed-tools" => {
let value = args let value = args
.get(index + 1) .get(index + 1)
@@ -428,15 +450,26 @@ fn print_bootstrap_plan() {
} }
} }
fn default_oauth_config() -> OAuthConfig {
OAuthConfig {
client_id: String::from("9d1c250a-e61b-44d9-88ed-5944d1962f5e"),
authorize_url: String::from("https://platform.claude.com/oauth/authorize"),
token_url: String::from("https://platform.claude.com/v1/oauth/token"),
callback_port: None,
manual_redirect_url: None,
scopes: vec![
String::from("user:profile"),
String::from("user:inference"),
String::from("user:sessions:claude_code"),
],
}
}
fn run_login() -> Result<(), Box<dyn std::error::Error>> { fn run_login() -> Result<(), Box<dyn std::error::Error>> {
let cwd = env::current_dir()?; let cwd = env::current_dir()?;
let config = ConfigLoader::default_for(&cwd).load()?; let config = ConfigLoader::default_for(&cwd).load()?;
let oauth = config.oauth().ok_or_else(|| { let default_oauth = default_oauth_config();
io::Error::new( let oauth = config.oauth().unwrap_or(&default_oauth);
io::ErrorKind::NotFound,
"OAuth config is missing. Add settings.oauth.clientId/authorizeUrl/tokenUrl first.",
)
})?;
let callback_port = oauth.callback_port.unwrap_or(DEFAULT_OAUTH_CALLBACK_PORT); let callback_port = oauth.callback_port.unwrap_or(DEFAULT_OAUTH_CALLBACK_PORT);
let redirect_uri = runtime::loopback_redirect_uri(callback_port); let redirect_uri = runtime::loopback_redirect_uri(callback_port);
let pkce = generate_pkce_pair()?; let pkce = generate_pkce_pair()?;
@@ -965,6 +998,7 @@ impl LiveCli {
let session = create_managed_session_handle()?; let session = create_managed_session_handle()?;
let runtime = build_runtime( let runtime = build_runtime(
Session::new(), Session::new(),
&session.id,
model.clone(), model.clone(),
system_prompt.clone(), system_prompt.clone(),
enable_tools, enable_tools,
@@ -1056,6 +1090,7 @@ impl LiveCli {
let session = self.runtime.session().clone(); let session = self.runtime.session().clone();
let mut runtime = build_runtime( let mut runtime = build_runtime(
session, session,
&self.session.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1202,6 +1237,7 @@ impl LiveCli {
let message_count = session.messages.len(); let message_count = session.messages.len();
self.runtime = build_runtime( self.runtime = build_runtime(
session, session,
&self.session.id,
model.clone(), model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1245,6 +1281,7 @@ impl LiveCli {
self.permission_mode = permission_mode_from_label(normalized); self.permission_mode = permission_mode_from_label(normalized);
self.runtime = build_runtime( self.runtime = build_runtime(
session, session,
&self.session.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1270,6 +1307,7 @@ impl LiveCli {
self.session = create_managed_session_handle()?; self.session = create_managed_session_handle()?;
self.runtime = build_runtime( self.runtime = build_runtime(
Session::new(), Session::new(),
&self.session.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1305,6 +1343,7 @@ impl LiveCli {
let message_count = session.messages.len(); let message_count = session.messages.len();
self.runtime = build_runtime( self.runtime = build_runtime(
session, session,
&handle.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1377,6 +1416,7 @@ impl LiveCli {
let message_count = session.messages.len(); let message_count = session.messages.len();
self.runtime = build_runtime( self.runtime = build_runtime(
session, session,
&handle.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1407,6 +1447,7 @@ impl LiveCli {
let skipped = removed == 0; let skipped = removed == 0;
self.runtime = build_runtime( self.runtime = build_runtime(
result.compacted_session, result.compacted_session,
&self.session.id,
self.model.clone(), self.model.clone(),
self.system_prompt.clone(), self.system_prompt.clone(),
true, true,
@@ -1873,8 +1914,19 @@ fn build_system_prompt() -> Result<Vec<String>, Box<dyn std::error::Error>> {
)?) )?)
} }
fn build_runtime_feature_config(
) -> Result<runtime::RuntimeFeatureConfig, Box<dyn std::error::Error>> {
let cwd = env::current_dir()?;
Ok(ConfigLoader::default_for(cwd)
.load()?
.feature_config()
.clone())
}
#[allow(clippy::too_many_arguments)]
fn build_runtime( fn build_runtime(
session: Session, session: Session,
session_id: &str,
model: String, model: String,
system_prompt: Vec<String>, system_prompt: Vec<String>,
enable_tools: bool, enable_tools: bool,
@@ -1883,13 +1935,41 @@ fn build_runtime(
permission_mode: PermissionMode, permission_mode: PermissionMode,
) -> Result<ConversationRuntime<AnthropicRuntimeClient, CliToolExecutor>, Box<dyn std::error::Error>> ) -> Result<ConversationRuntime<AnthropicRuntimeClient, CliToolExecutor>, Box<dyn std::error::Error>>
{ {
Ok(ConversationRuntime::new( let session_tracer = build_session_tracer(session_id)?;
let api_client = match session_tracer.clone() {
Some(session_tracer) => {
AnthropicRuntimeClient::new(model, enable_tools, emit_output, allowed_tools.clone())?
.with_session_tracer(session_tracer)
}
None => {
AnthropicRuntimeClient::new(model, enable_tools, emit_output, allowed_tools.clone())?
}
};
let runtime = ConversationRuntime::new_with_features(
session, session,
AnthropicRuntimeClient::new(model, enable_tools, emit_output, allowed_tools.clone())?, api_client,
CliToolExecutor::new(allowed_tools, emit_output), CliToolExecutor::new(allowed_tools, emit_output),
permission_policy(permission_mode), permission_policy(permission_mode),
system_prompt, system_prompt,
)) &build_runtime_feature_config()?,
);
Ok(match session_tracer {
Some(session_tracer) => runtime.with_session_tracer(session_tracer),
None => runtime,
})
}
fn build_session_tracer(
session_id: &str,
) -> Result<Option<SessionTracer>, Box<dyn std::error::Error>> {
let Some(path) = env::var_os(TELEMETRY_LOG_PATH_ENV) else {
return Ok(None);
};
let sink = JsonlTelemetrySink::new(PathBuf::from(path))?;
Ok(Some(SessionTracer::new(
session_id.to_string(),
std::sync::Arc::new(sink),
)))
} }
struct CliPermissionPrompter { struct CliPermissionPrompter {
@@ -1964,6 +2044,11 @@ impl AnthropicRuntimeClient {
allowed_tools, allowed_tools,
}) })
} }
fn with_session_tracer(mut self, session_tracer: SessionTracer) -> Self {
self.client = self.client.with_session_tracer(session_tracer);
self
}
} }
fn resolve_cli_auth_source() -> Result<AuthSource, Box<dyn std::error::Error>> { fn resolve_cli_auth_source() -> Result<AuthSource, Box<dyn std::error::Error>> {
@@ -2069,12 +2154,7 @@ impl ApiClient for AnthropicRuntimeClient {
} }
} }
ApiStreamEvent::MessageDelta(delta) => { ApiStreamEvent::MessageDelta(delta) => {
events.push(AssistantEvent::Usage(TokenUsage { events.push(AssistantEvent::Usage(delta.usage.token_usage()));
input_tokens: delta.usage.input_tokens,
output_tokens: delta.usage.output_tokens,
cache_creation_input_tokens: 0,
cache_read_input_tokens: 0,
}));
} }
ApiStreamEvent::MessageStop(_) => { ApiStreamEvent::MessageStop(_) => {
saw_stop = true; saw_stop = true;
@@ -2324,13 +2404,13 @@ fn format_bash_result(icon: &str, parsed: &serde_json::Value) -> String {
.get("backgroundTaskId") .get("backgroundTaskId")
.and_then(|value| value.as_str()) .and_then(|value| value.as_str())
{ {
lines[0].push_str(&format!(" backgrounded ({task_id})")); write!(&mut lines[0], " backgrounded ({task_id})").expect("write to string");
} else if let Some(status) = parsed } else if let Some(status) = parsed
.get("returnCodeInterpretation") .get("returnCodeInterpretation")
.and_then(|value| value.as_str()) .and_then(|value| value.as_str())
.filter(|status| !status.is_empty()) .filter(|status| !status.is_empty())
{ {
lines[0].push_str(&format!(" {status}")); write!(&mut lines[0], " {status}").expect("write to string");
} }
if let Some(stdout) = parsed.get("stdout").and_then(|value| value.as_str()) { if let Some(stdout) = parsed.get("stdout").and_then(|value| value.as_str()) {
@@ -2352,15 +2432,15 @@ fn format_read_result(icon: &str, parsed: &serde_json::Value) -> String {
let path = extract_tool_path(file); let path = extract_tool_path(file);
let start_line = file let start_line = file
.get("startLine") .get("startLine")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(1); .unwrap_or(1);
let num_lines = file let num_lines = file
.get("numLines") .get("numLines")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(0); .unwrap_or(0);
let total_lines = file let total_lines = file
.get("totalLines") .get("totalLines")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(num_lines); .unwrap_or(num_lines);
let content = file let content = file
.get("content") .get("content")
@@ -2386,8 +2466,7 @@ fn format_write_result(icon: &str, parsed: &serde_json::Value) -> String {
let line_count = parsed let line_count = parsed
.get("content") .get("content")
.and_then(|value| value.as_str()) .and_then(|value| value.as_str())
.map(|content| content.lines().count()) .map_or(0, |content| content.lines().count());
.unwrap_or(0);
format!( format!(
"{icon} \x1b[1;32m✏ {} {path}\x1b[0m \x1b[2m({line_count} lines)\x1b[0m", "{icon} \x1b[1;32m✏ {} {path}\x1b[0m \x1b[2m({line_count} lines)\x1b[0m",
if kind == "create" { "Wrote" } else { "Updated" }, if kind == "create" { "Wrote" } else { "Updated" },
@@ -2418,7 +2497,7 @@ fn format_edit_result(icon: &str, parsed: &serde_json::Value) -> String {
let path = extract_tool_path(parsed); let path = extract_tool_path(parsed);
let suffix = if parsed let suffix = if parsed
.get("replaceAll") .get("replaceAll")
.and_then(|value| value.as_bool()) .and_then(serde_json::Value::as_bool)
.unwrap_or(false) .unwrap_or(false)
{ {
" (replace all)" " (replace all)"
@@ -2446,7 +2525,7 @@ fn format_edit_result(icon: &str, parsed: &serde_json::Value) -> String {
fn format_glob_result(icon: &str, parsed: &serde_json::Value) -> String { fn format_glob_result(icon: &str, parsed: &serde_json::Value) -> String {
let num_files = parsed let num_files = parsed
.get("numFiles") .get("numFiles")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(0); .unwrap_or(0);
let filenames = parsed let filenames = parsed
.get("filenames") .get("filenames")
@@ -2470,11 +2549,11 @@ fn format_glob_result(icon: &str, parsed: &serde_json::Value) -> String {
fn format_grep_result(icon: &str, parsed: &serde_json::Value) -> String { fn format_grep_result(icon: &str, parsed: &serde_json::Value) -> String {
let num_matches = parsed let num_matches = parsed
.get("numMatches") .get("numMatches")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(0); .unwrap_or(0);
let num_files = parsed let num_files = parsed
.get("numFiles") .get("numFiles")
.and_then(|value| value.as_u64()) .and_then(serde_json::Value::as_u64)
.unwrap_or(0); .unwrap_or(0);
let content = parsed let content = parsed
.get("content") .get("content")
@@ -2571,12 +2650,7 @@ fn response_to_events(
} }
} }
events.push(AssistantEvent::Usage(TokenUsage { events.push(AssistantEvent::Usage(response.usage.token_usage()));
input_tokens: response.usage.input_tokens,
output_tokens: response.usage.output_tokens,
cache_creation_input_tokens: response.usage.cache_creation_input_tokens,
cache_read_input_tokens: response.usage.cache_read_input_tokens,
}));
events.push(AssistantEvent::MessageStop); events.push(AssistantEvent::MessageStop);
Ok(events) Ok(events)
} }

View File

@@ -286,7 +286,7 @@ impl TerminalRenderer {
) { ) {
match event { match event {
Event::Start(Tag::Heading { level, .. }) => { Event::Start(Tag::Heading { level, .. }) => {
self.start_heading(state, level as u8, output) Self::start_heading(state, level as u8, output);
} }
Event::End(TagEnd::Paragraph) => output.push_str("\n\n"), Event::End(TagEnd::Paragraph) => output.push_str("\n\n"),
Event::Start(Tag::BlockQuote(..)) => self.start_quote(state, output), Event::Start(Tag::BlockQuote(..)) => self.start_quote(state, output),
@@ -426,7 +426,7 @@ impl TerminalRenderer {
} }
} }
fn start_heading(&self, state: &mut RenderState, level: u8, output: &mut String) { fn start_heading(state: &mut RenderState, level: u8, output: &mut String) {
state.heading_level = Some(level); state.heading_level = Some(level);
if !output.is_empty() { if !output.is_empty() {
output.push('\n'); output.push('\n');

View File

@@ -0,0 +1,13 @@
[package]
name = "telemetry"
version.workspace = true
edition.workspace = true
license.workspace = true
publish.workspace = true
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
[lints]
workspace = true

View File

@@ -0,0 +1,526 @@
use std::fmt::{Debug, Formatter};
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
pub const DEFAULT_ANTHROPIC_VERSION: &str = "2023-06-01";
pub const DEFAULT_APP_NAME: &str = "claude-code";
pub const DEFAULT_RUNTIME: &str = "rust";
pub const DEFAULT_AGENTIC_BETA: &str = "claude-code-20250219";
pub const DEFAULT_PROMPT_CACHING_SCOPE_BETA: &str = "prompt-caching-scope-2026-01-05";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientIdentity {
pub app_name: String,
pub app_version: String,
pub runtime: String,
}
impl ClientIdentity {
#[must_use]
pub fn new(app_name: impl Into<String>, app_version: impl Into<String>) -> Self {
Self {
app_name: app_name.into(),
app_version: app_version.into(),
runtime: DEFAULT_RUNTIME.to_string(),
}
}
#[must_use]
pub fn with_runtime(mut self, runtime: impl Into<String>) -> Self {
self.runtime = runtime.into();
self
}
#[must_use]
pub fn user_agent(&self) -> String {
format!("{}/{}", self.app_name, self.app_version)
}
}
impl Default for ClientIdentity {
fn default() -> Self {
Self::new(DEFAULT_APP_NAME, env!("CARGO_PKG_VERSION"))
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AnthropicRequestProfile {
pub anthropic_version: String,
pub client_identity: ClientIdentity,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub betas: Vec<String>,
#[serde(default, skip_serializing_if = "Map::is_empty")]
pub extra_body: Map<String, Value>,
}
impl AnthropicRequestProfile {
#[must_use]
pub fn new(client_identity: ClientIdentity) -> Self {
Self {
anthropic_version: DEFAULT_ANTHROPIC_VERSION.to_string(),
client_identity,
betas: vec![
DEFAULT_AGENTIC_BETA.to_string(),
DEFAULT_PROMPT_CACHING_SCOPE_BETA.to_string(),
],
extra_body: Map::new(),
}
}
#[must_use]
pub fn with_beta(mut self, beta: impl Into<String>) -> Self {
let beta = beta.into();
if !self.betas.contains(&beta) {
self.betas.push(beta);
}
self
}
#[must_use]
pub fn with_extra_body(mut self, key: impl Into<String>, value: Value) -> Self {
self.extra_body.insert(key.into(), value);
self
}
#[must_use]
pub fn header_pairs(&self) -> Vec<(String, String)> {
let mut headers = vec![
(
"anthropic-version".to_string(),
self.anthropic_version.clone(),
),
("user-agent".to_string(), self.client_identity.user_agent()),
];
if !self.betas.is_empty() {
headers.push(("anthropic-beta".to_string(), self.betas.join(",")));
}
headers
}
pub fn render_json_body<T: Serialize>(&self, request: &T) -> Result<Value, serde_json::Error> {
let mut body = serde_json::to_value(request)?;
let object = body.as_object_mut().ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"request body must serialize to a JSON object",
))
})?;
for (key, value) in &self.extra_body {
object.insert(key.clone(), value.clone());
}
if !self.betas.is_empty() {
object.insert(
"betas".to_string(),
Value::Array(self.betas.iter().cloned().map(Value::String).collect()),
);
}
Ok(body)
}
}
impl Default for AnthropicRequestProfile {
fn default() -> Self {
Self::new(ClientIdentity::default())
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AnalyticsEvent {
pub namespace: String,
pub action: String,
#[serde(default, skip_serializing_if = "Map::is_empty")]
pub properties: Map<String, Value>,
}
impl AnalyticsEvent {
#[must_use]
pub fn new(namespace: impl Into<String>, action: impl Into<String>) -> Self {
Self {
namespace: namespace.into(),
action: action.into(),
properties: Map::new(),
}
}
#[must_use]
pub fn with_property(mut self, key: impl Into<String>, value: Value) -> Self {
self.properties.insert(key.into(), value);
self
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SessionTraceRecord {
pub session_id: String,
pub sequence: u64,
pub name: String,
pub timestamp_ms: u64,
#[serde(default, skip_serializing_if = "Map::is_empty")]
pub attributes: Map<String, Value>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum TelemetryEvent {
HttpRequestStarted {
session_id: String,
attempt: u32,
method: String,
path: String,
#[serde(default, skip_serializing_if = "Map::is_empty")]
attributes: Map<String, Value>,
},
HttpRequestSucceeded {
session_id: String,
attempt: u32,
method: String,
path: String,
status: u16,
#[serde(default, skip_serializing_if = "Option::is_none")]
request_id: Option<String>,
#[serde(default, skip_serializing_if = "Map::is_empty")]
attributes: Map<String, Value>,
},
HttpRequestFailed {
session_id: String,
attempt: u32,
method: String,
path: String,
error: String,
retryable: bool,
#[serde(default, skip_serializing_if = "Map::is_empty")]
attributes: Map<String, Value>,
},
Analytics(AnalyticsEvent),
SessionTrace(SessionTraceRecord),
}
pub trait TelemetrySink: Send + Sync {
fn record(&self, event: TelemetryEvent);
}
#[derive(Default)]
pub struct MemoryTelemetrySink {
events: Mutex<Vec<TelemetryEvent>>,
}
impl MemoryTelemetrySink {
#[must_use]
pub fn events(&self) -> Vec<TelemetryEvent> {
self.events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
}
impl TelemetrySink for MemoryTelemetrySink {
fn record(&self, event: TelemetryEvent) {
self.events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(event);
}
}
pub struct JsonlTelemetrySink {
path: PathBuf,
file: Mutex<File>,
}
impl Debug for JsonlTelemetrySink {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.debug_struct("JsonlTelemetrySink")
.field("path", &self.path)
.finish_non_exhaustive()
}
}
impl JsonlTelemetrySink {
pub fn new(path: impl AsRef<Path>) -> Result<Self, std::io::Error> {
let path = path.as_ref().to_path_buf();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let file = OpenOptions::new().create(true).append(true).open(&path)?;
Ok(Self {
path,
file: Mutex::new(file),
})
}
#[must_use]
pub fn path(&self) -> &Path {
&self.path
}
}
impl TelemetrySink for JsonlTelemetrySink {
fn record(&self, event: TelemetryEvent) {
let Ok(line) = serde_json::to_string(&event) else {
return;
};
let mut file = self
.file
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let _ = writeln!(file, "{line}");
let _ = file.flush();
}
}
#[derive(Clone)]
pub struct SessionTracer {
session_id: String,
sequence: Arc<AtomicU64>,
sink: Arc<dyn TelemetrySink>,
}
impl Debug for SessionTracer {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SessionTracer")
.field("session_id", &self.session_id)
.finish_non_exhaustive()
}
}
impl SessionTracer {
#[must_use]
pub fn new(session_id: impl Into<String>, sink: Arc<dyn TelemetrySink>) -> Self {
Self {
session_id: session_id.into(),
sequence: Arc::new(AtomicU64::new(0)),
sink,
}
}
#[must_use]
pub fn session_id(&self) -> &str {
&self.session_id
}
pub fn record(&self, name: impl Into<String>, attributes: Map<String, Value>) {
let record = SessionTraceRecord {
session_id: self.session_id.clone(),
sequence: self.sequence.fetch_add(1, Ordering::Relaxed),
name: name.into(),
timestamp_ms: current_timestamp_ms(),
attributes,
};
self.sink.record(TelemetryEvent::SessionTrace(record));
}
pub fn record_http_request_started(
&self,
attempt: u32,
method: impl Into<String>,
path: impl Into<String>,
attributes: Map<String, Value>,
) {
let method = method.into();
let path = path.into();
self.sink.record(TelemetryEvent::HttpRequestStarted {
session_id: self.session_id.clone(),
attempt,
method: method.clone(),
path: path.clone(),
attributes: attributes.clone(),
});
self.record(
"http_request_started",
merge_trace_fields(method, path, attempt, attributes),
);
}
pub fn record_http_request_succeeded(
&self,
attempt: u32,
method: impl Into<String>,
path: impl Into<String>,
status: u16,
request_id: Option<String>,
attributes: Map<String, Value>,
) {
let method = method.into();
let path = path.into();
self.sink.record(TelemetryEvent::HttpRequestSucceeded {
session_id: self.session_id.clone(),
attempt,
method: method.clone(),
path: path.clone(),
status,
request_id: request_id.clone(),
attributes: attributes.clone(),
});
let mut trace_attributes = merge_trace_fields(method, path, attempt, attributes);
trace_attributes.insert("status".to_string(), Value::from(status));
if let Some(request_id) = request_id {
trace_attributes.insert("request_id".to_string(), Value::String(request_id));
}
self.record("http_request_succeeded", trace_attributes);
}
pub fn record_http_request_failed(
&self,
attempt: u32,
method: impl Into<String>,
path: impl Into<String>,
error: impl Into<String>,
retryable: bool,
attributes: Map<String, Value>,
) {
let method = method.into();
let path = path.into();
let error = error.into();
self.sink.record(TelemetryEvent::HttpRequestFailed {
session_id: self.session_id.clone(),
attempt,
method: method.clone(),
path: path.clone(),
error: error.clone(),
retryable,
attributes: attributes.clone(),
});
let mut trace_attributes = merge_trace_fields(method, path, attempt, attributes);
trace_attributes.insert("error".to_string(), Value::String(error));
trace_attributes.insert("retryable".to_string(), Value::Bool(retryable));
self.record("http_request_failed", trace_attributes);
}
pub fn record_analytics(&self, event: AnalyticsEvent) {
let mut attributes = event.properties.clone();
attributes.insert(
"namespace".to_string(),
Value::String(event.namespace.clone()),
);
attributes.insert("action".to_string(), Value::String(event.action.clone()));
self.sink.record(TelemetryEvent::Analytics(event));
self.record("analytics", attributes);
}
}
fn merge_trace_fields(
method: String,
path: String,
attempt: u32,
mut attributes: Map<String, Value>,
) -> Map<String, Value> {
attributes.insert("method".to_string(), Value::String(method));
attributes.insert("path".to_string(), Value::String(path));
attributes.insert("attempt".to_string(), Value::from(attempt));
attributes
}
fn current_timestamp_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn request_profile_emits_headers_and_merges_body() {
let profile = AnthropicRequestProfile::new(
ClientIdentity::new("claude-code", "1.2.3").with_runtime("rust-cli"),
)
.with_beta("tools-2026-04-01")
.with_extra_body("metadata", serde_json::json!({"source": "test"}));
assert_eq!(
profile.header_pairs(),
vec![
(
"anthropic-version".to_string(),
DEFAULT_ANTHROPIC_VERSION.to_string()
),
("user-agent".to_string(), "claude-code/1.2.3".to_string()),
(
"anthropic-beta".to_string(),
"claude-code-20250219,prompt-caching-scope-2026-01-05,tools-2026-04-01"
.to_string(),
),
]
);
let body = profile
.render_json_body(&serde_json::json!({"model": "claude-sonnet"}))
.expect("body should serialize");
assert_eq!(
body["metadata"]["source"],
Value::String("test".to_string())
);
assert_eq!(
body["betas"],
serde_json::json!([
"claude-code-20250219",
"prompt-caching-scope-2026-01-05",
"tools-2026-04-01"
])
);
}
#[test]
fn session_tracer_records_structured_events_and_trace_sequence() {
let sink = Arc::new(MemoryTelemetrySink::default());
let tracer = SessionTracer::new("session-123", sink.clone());
tracer.record_http_request_started(1, "POST", "/v1/messages", Map::new());
tracer.record_analytics(
AnalyticsEvent::new("cli", "prompt_sent")
.with_property("model", Value::String("claude-opus".to_string())),
);
let events = sink.events();
assert!(matches!(
&events[0],
TelemetryEvent::HttpRequestStarted {
session_id,
attempt: 1,
method,
path,
..
} if session_id == "session-123" && method == "POST" && path == "/v1/messages"
));
assert!(matches!(
&events[1],
TelemetryEvent::SessionTrace(SessionTraceRecord { sequence: 0, name, .. })
if name == "http_request_started"
));
assert!(matches!(&events[2], TelemetryEvent::Analytics(_)));
assert!(matches!(
&events[3],
TelemetryEvent::SessionTrace(SessionTraceRecord { sequence: 1, name, .. })
if name == "analytics"
));
}
#[test]
fn jsonl_sink_persists_events() {
let path =
std::env::temp_dir().join(format!("telemetry-jsonl-{}.log", current_timestamp_ms()));
let sink = JsonlTelemetrySink::new(&path).expect("sink should create file");
sink.record(TelemetryEvent::Analytics(
AnalyticsEvent::new("cli", "turn_completed").with_property("ok", Value::Bool(true)),
));
let contents = std::fs::read_to_string(&path).expect("telemetry log should be readable");
assert!(contents.contains("\"type\":\"analytics\""));
assert!(contents.contains("\"action\":\"turn_completed\""));
let _ = std::fs::remove_file(path);
}
}

View File

@@ -6,10 +6,12 @@ license.workspace = true
publish.workspace = true publish.workspace = true
[dependencies] [dependencies]
api = { path = "../api" }
runtime = { path = "../runtime" } runtime = { path = "../runtime" }
reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls"] } reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls"] }
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
tokio = { version = "1", features = ["rt-multi-thread"] }
[lints] [lints]
workspace = true workspace = true

View File

@@ -3,10 +3,17 @@ use std::path::{Path, PathBuf};
use std::process::Command; use std::process::Command;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use api::{
read_base_url, AnthropicClient, ContentBlockDelta, InputContentBlock, InputMessage,
MessageRequest, MessageResponse, OutputContentBlock, StreamEvent as ApiStreamEvent, ToolChoice,
ToolDefinition, ToolResultContentBlock,
};
use reqwest::blocking::Client; use reqwest::blocking::Client;
use runtime::{ use runtime::{
edit_file, execute_bash, glob_search, grep_search, read_file, write_file, BashCommandInput, edit_file, execute_bash, glob_search, grep_search, load_system_prompt, read_file, write_file,
GrepSearchInput, PermissionMode, ApiClient, ApiRequest, AssistantEvent, BashCommandInput, ContentBlock, ConversationMessage,
ConversationRuntime, GrepSearchInput, MessageRole, PermissionMode, PermissionPolicy,
RuntimeError, Session, ToolError, ToolExecutor,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::{json, Value}; use serde_json::{json, Value};
@@ -702,7 +709,7 @@ struct SkillOutput {
prompt: String, prompt: String,
} }
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
struct AgentOutput { struct AgentOutput {
#[serde(rename = "agentId")] #[serde(rename = "agentId")]
agent_id: String, agent_id: String,
@@ -718,6 +725,20 @@ struct AgentOutput {
manifest_file: String, manifest_file: String,
#[serde(rename = "createdAt")] #[serde(rename = "createdAt")]
created_at: String, created_at: String,
#[serde(rename = "startedAt", skip_serializing_if = "Option::is_none")]
started_at: Option<String>,
#[serde(rename = "completedAt", skip_serializing_if = "Option::is_none")]
completed_at: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
}
#[derive(Debug, Clone)]
struct AgentJob {
manifest: AgentOutput,
prompt: String,
system_prompt: Vec<String>,
allowed_tools: BTreeSet<String>,
} }
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
@@ -1315,7 +1336,18 @@ fn resolve_skill_path(skill: &str) -> Result<std::path::PathBuf, String> {
Err(format!("unknown skill: {requested}")) Err(format!("unknown skill: {requested}"))
} }
const DEFAULT_AGENT_MODEL: &str = "claude-opus-4-6";
const DEFAULT_AGENT_SYSTEM_DATE: &str = "2026-03-31";
const DEFAULT_AGENT_MAX_ITERATIONS: usize = 32;
fn execute_agent(input: AgentInput) -> Result<AgentOutput, String> { fn execute_agent(input: AgentInput) -> Result<AgentOutput, String> {
execute_agent_with_spawn(input, spawn_agent_job)
}
fn execute_agent_with_spawn<F>(input: AgentInput, spawn_fn: F) -> Result<AgentOutput, String>
where
F: FnOnce(AgentJob) -> Result<(), String>,
{
if input.description.trim().is_empty() { if input.description.trim().is_empty() {
return Err(String::from("description must not be empty")); return Err(String::from("description must not be empty"));
} }
@@ -1329,6 +1361,7 @@ fn execute_agent(input: AgentInput) -> Result<AgentOutput, String> {
let output_file = output_dir.join(format!("{agent_id}.md")); let output_file = output_dir.join(format!("{agent_id}.md"));
let manifest_file = output_dir.join(format!("{agent_id}.json")); let manifest_file = output_dir.join(format!("{agent_id}.json"));
let normalized_subagent_type = normalize_subagent_type(input.subagent_type.as_deref()); let normalized_subagent_type = normalize_subagent_type(input.subagent_type.as_deref());
let model = resolve_agent_model(input.model.as_deref());
let agent_name = input let agent_name = input
.name .name
.as_deref() .as_deref()
@@ -1336,6 +1369,8 @@ fn execute_agent(input: AgentInput) -> Result<AgentOutput, String> {
.filter(|name| !name.is_empty()) .filter(|name| !name.is_empty())
.unwrap_or_else(|| slugify_agent_name(&input.description)); .unwrap_or_else(|| slugify_agent_name(&input.description));
let created_at = iso8601_now(); let created_at = iso8601_now();
let system_prompt = build_agent_system_prompt(&normalized_subagent_type)?;
let allowed_tools = allowed_tools_for_subagent(&normalized_subagent_type);
let output_contents = format!( let output_contents = format!(
"# Agent Task "# Agent Task
@@ -1359,21 +1394,504 @@ fn execute_agent(input: AgentInput) -> Result<AgentOutput, String> {
name: agent_name, name: agent_name,
description: input.description, description: input.description,
subagent_type: Some(normalized_subagent_type), subagent_type: Some(normalized_subagent_type),
model: input.model, model: Some(model),
status: String::from("queued"), status: String::from("running"),
output_file: output_file.display().to_string(), output_file: output_file.display().to_string(),
manifest_file: manifest_file.display().to_string(), manifest_file: manifest_file.display().to_string(),
created_at, created_at: created_at.clone(),
started_at: Some(created_at),
completed_at: None,
error: None,
}; };
std::fs::write( write_agent_manifest(&manifest)?;
&manifest_file,
serde_json::to_string_pretty(&manifest).map_err(|error| error.to_string())?, let manifest_for_spawn = manifest.clone();
) let job = AgentJob {
.map_err(|error| error.to_string())?; manifest: manifest_for_spawn,
prompt: input.prompt,
system_prompt,
allowed_tools,
};
if let Err(error) = spawn_fn(job) {
let error = format!("failed to spawn sub-agent: {error}");
persist_agent_terminal_state(&manifest, "failed", None, Some(error.clone()))?;
return Err(error);
}
Ok(manifest) Ok(manifest)
} }
fn spawn_agent_job(job: AgentJob) -> Result<(), String> {
let thread_name = format!("clawd-agent-{}", job.manifest.agent_id);
std::thread::Builder::new()
.name(thread_name)
.spawn(move || {
let result =
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run_agent_job(&job)));
match result {
Ok(Ok(())) => {}
Ok(Err(error)) => {
let _ =
persist_agent_terminal_state(&job.manifest, "failed", None, Some(error));
}
Err(_) => {
let _ = persist_agent_terminal_state(
&job.manifest,
"failed",
None,
Some(String::from("sub-agent thread panicked")),
);
}
}
})
.map(|_| ())
.map_err(|error| error.to_string())
}
fn run_agent_job(job: &AgentJob) -> Result<(), String> {
let mut runtime = build_agent_runtime(job)?.with_max_iterations(DEFAULT_AGENT_MAX_ITERATIONS);
let summary = runtime
.run_turn(job.prompt.clone(), None)
.map_err(|error| error.to_string())?;
let final_text = final_assistant_text(&summary);
persist_agent_terminal_state(&job.manifest, "completed", Some(final_text.as_str()), None)
}
fn build_agent_runtime(
job: &AgentJob,
) -> Result<ConversationRuntime<AnthropicRuntimeClient, SubagentToolExecutor>, String> {
let model = job
.manifest
.model
.clone()
.unwrap_or_else(|| DEFAULT_AGENT_MODEL.to_string());
let allowed_tools = job.allowed_tools.clone();
let api_client = AnthropicRuntimeClient::new(model, allowed_tools.clone())?;
let tool_executor = SubagentToolExecutor::new(allowed_tools);
Ok(ConversationRuntime::new(
Session::new(),
api_client,
tool_executor,
agent_permission_policy(),
job.system_prompt.clone(),
))
}
fn build_agent_system_prompt(subagent_type: &str) -> Result<Vec<String>, String> {
let cwd = std::env::current_dir().map_err(|error| error.to_string())?;
let mut prompt = load_system_prompt(
cwd,
DEFAULT_AGENT_SYSTEM_DATE.to_string(),
std::env::consts::OS,
"unknown",
)
.map_err(|error| error.to_string())?;
prompt.push(format!(
"You are a background sub-agent of type `{subagent_type}`. Work only on the delegated task, use only the tools available to you, do not ask the user questions, and finish with a concise result."
));
Ok(prompt)
}
fn resolve_agent_model(model: Option<&str>) -> String {
model
.map(str::trim)
.filter(|model| !model.is_empty())
.unwrap_or(DEFAULT_AGENT_MODEL)
.to_string()
}
fn allowed_tools_for_subagent(subagent_type: &str) -> BTreeSet<String> {
let tools = match subagent_type {
"Explore" => vec![
"read_file",
"glob_search",
"grep_search",
"WebFetch",
"WebSearch",
"ToolSearch",
"Skill",
"StructuredOutput",
],
"Plan" => vec![
"read_file",
"glob_search",
"grep_search",
"WebFetch",
"WebSearch",
"ToolSearch",
"Skill",
"TodoWrite",
"StructuredOutput",
"SendUserMessage",
],
"Verification" => vec![
"bash",
"read_file",
"glob_search",
"grep_search",
"WebFetch",
"WebSearch",
"ToolSearch",
"TodoWrite",
"StructuredOutput",
"SendUserMessage",
"PowerShell",
],
"claude-code-guide" => vec![
"read_file",
"glob_search",
"grep_search",
"WebFetch",
"WebSearch",
"ToolSearch",
"Skill",
"StructuredOutput",
"SendUserMessage",
],
"statusline-setup" => vec![
"bash",
"read_file",
"write_file",
"edit_file",
"glob_search",
"grep_search",
"ToolSearch",
],
_ => vec![
"bash",
"read_file",
"write_file",
"edit_file",
"glob_search",
"grep_search",
"WebFetch",
"WebSearch",
"TodoWrite",
"Skill",
"ToolSearch",
"NotebookEdit",
"Sleep",
"SendUserMessage",
"Config",
"StructuredOutput",
"REPL",
"PowerShell",
],
};
tools.into_iter().map(str::to_string).collect()
}
fn agent_permission_policy() -> PermissionPolicy {
mvp_tool_specs().into_iter().fold(
PermissionPolicy::new(PermissionMode::DangerFullAccess),
|policy, spec| policy.with_tool_requirement(spec.name, spec.required_permission),
)
}
fn write_agent_manifest(manifest: &AgentOutput) -> Result<(), String> {
std::fs::write(
&manifest.manifest_file,
serde_json::to_string_pretty(manifest).map_err(|error| error.to_string())?,
)
.map_err(|error| error.to_string())
}
fn persist_agent_terminal_state(
manifest: &AgentOutput,
status: &str,
result: Option<&str>,
error: Option<String>,
) -> Result<(), String> {
append_agent_output(
&manifest.output_file,
&format_agent_terminal_output(status, result, error.as_deref()),
)?;
let mut next_manifest = manifest.clone();
next_manifest.status = status.to_string();
next_manifest.completed_at = Some(iso8601_now());
next_manifest.error = error;
write_agent_manifest(&next_manifest)
}
fn append_agent_output(path: &str, suffix: &str) -> Result<(), String> {
use std::io::Write as _;
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(path)
.map_err(|error| error.to_string())?;
file.write_all(suffix.as_bytes())
.map_err(|error| error.to_string())
}
fn format_agent_terminal_output(status: &str, result: Option<&str>, error: Option<&str>) -> String {
let mut sections = vec![format!("\n## Result\n\n- status: {status}\n")];
if let Some(result) = result.filter(|value| !value.trim().is_empty()) {
sections.push(format!("\n### Final response\n\n{}\n", result.trim()));
}
if let Some(error) = error.filter(|value| !value.trim().is_empty()) {
sections.push(format!("\n### Error\n\n{}\n", error.trim()));
}
sections.join("")
}
struct AnthropicRuntimeClient {
runtime: tokio::runtime::Runtime,
client: AnthropicClient,
model: String,
allowed_tools: BTreeSet<String>,
}
impl AnthropicRuntimeClient {
fn new(model: String, allowed_tools: BTreeSet<String>) -> Result<Self, String> {
let client = AnthropicClient::from_env()
.map_err(|error| error.to_string())?
.with_base_url(read_base_url());
Ok(Self {
runtime: tokio::runtime::Runtime::new().map_err(|error| error.to_string())?,
client,
model,
allowed_tools,
})
}
}
impl ApiClient for AnthropicRuntimeClient {
fn stream(&mut self, request: ApiRequest) -> Result<Vec<AssistantEvent>, RuntimeError> {
let tools = tool_specs_for_allowed_tools(Some(&self.allowed_tools))
.into_iter()
.map(|spec| ToolDefinition {
name: spec.name.to_string(),
description: Some(spec.description.to_string()),
input_schema: spec.input_schema,
})
.collect::<Vec<_>>();
let message_request = MessageRequest {
model: self.model.clone(),
max_tokens: 32_000,
messages: convert_messages(&request.messages),
system: (!request.system_prompt.is_empty()).then(|| request.system_prompt.join("\n\n")),
tools: (!tools.is_empty()).then_some(tools),
tool_choice: (!self.allowed_tools.is_empty()).then_some(ToolChoice::Auto),
stream: true,
};
self.runtime.block_on(async {
let mut stream = self
.client
.stream_message(&message_request)
.await
.map_err(|error| RuntimeError::new(error.to_string()))?;
let mut events = Vec::new();
let mut pending_tool: Option<(String, String, String)> = None;
let mut saw_stop = false;
while let Some(event) = stream
.next_event()
.await
.map_err(|error| RuntimeError::new(error.to_string()))?
{
match event {
ApiStreamEvent::MessageStart(start) => {
for block in start.message.content {
push_output_block(block, &mut events, &mut pending_tool, true);
}
}
ApiStreamEvent::ContentBlockStart(start) => {
push_output_block(
start.content_block,
&mut events,
&mut pending_tool,
true,
);
}
ApiStreamEvent::ContentBlockDelta(delta) => match delta.delta {
ContentBlockDelta::TextDelta { text } => {
if !text.is_empty() {
events.push(AssistantEvent::TextDelta(text));
}
}
ContentBlockDelta::InputJsonDelta { partial_json } => {
if let Some((_, _, input)) = &mut pending_tool {
input.push_str(&partial_json);
}
}
},
ApiStreamEvent::ContentBlockStop(_) => {
if let Some((id, name, input)) = pending_tool.take() {
events.push(AssistantEvent::ToolUse { id, name, input });
}
}
ApiStreamEvent::MessageDelta(delta) => {
events.push(AssistantEvent::Usage(delta.usage.token_usage()));
}
ApiStreamEvent::MessageStop(_) => {
saw_stop = true;
events.push(AssistantEvent::MessageStop);
}
}
}
if !saw_stop
&& events.iter().any(|event| {
matches!(event, AssistantEvent::TextDelta(text) if !text.is_empty())
|| matches!(event, AssistantEvent::ToolUse { .. })
})
{
events.push(AssistantEvent::MessageStop);
}
if events
.iter()
.any(|event| matches!(event, AssistantEvent::MessageStop))
{
return Ok(events);
}
let response = self
.client
.send_message(&MessageRequest {
stream: false,
..message_request.clone()
})
.await
.map_err(|error| RuntimeError::new(error.to_string()))?;
Ok(response_to_events(response))
})
}
}
struct SubagentToolExecutor {
allowed_tools: BTreeSet<String>,
}
impl SubagentToolExecutor {
fn new(allowed_tools: BTreeSet<String>) -> Self {
Self { allowed_tools }
}
}
impl ToolExecutor for SubagentToolExecutor {
fn execute(&mut self, tool_name: &str, input: &str) -> Result<String, ToolError> {
if !self.allowed_tools.contains(tool_name) {
return Err(ToolError::new(format!(
"tool `{tool_name}` is not enabled for this sub-agent"
)));
}
let value = serde_json::from_str(input)
.map_err(|error| ToolError::new(format!("invalid tool input JSON: {error}")))?;
execute_tool(tool_name, &value).map_err(ToolError::new)
}
}
fn tool_specs_for_allowed_tools(allowed_tools: Option<&BTreeSet<String>>) -> Vec<ToolSpec> {
mvp_tool_specs()
.into_iter()
.filter(|spec| allowed_tools.is_none_or(|allowed| allowed.contains(spec.name)))
.collect()
}
fn convert_messages(messages: &[ConversationMessage]) -> Vec<InputMessage> {
messages
.iter()
.filter_map(|message| {
let role = match message.role {
MessageRole::System | MessageRole::User | MessageRole::Tool => "user",
MessageRole::Assistant => "assistant",
};
let content = message
.blocks
.iter()
.map(|block| match block {
ContentBlock::Text { text } => InputContentBlock::Text { text: text.clone() },
ContentBlock::ToolUse { id, name, input } => InputContentBlock::ToolUse {
id: id.clone(),
name: name.clone(),
input: serde_json::from_str(input)
.unwrap_or_else(|_| serde_json::json!({ "raw": input })),
},
ContentBlock::ToolResult {
tool_use_id,
output,
is_error,
..
} => InputContentBlock::ToolResult {
tool_use_id: tool_use_id.clone(),
content: vec![ToolResultContentBlock::Text {
text: output.clone(),
}],
is_error: *is_error,
},
})
.collect::<Vec<_>>();
(!content.is_empty()).then(|| InputMessage {
role: role.to_string(),
content,
})
})
.collect()
}
fn push_output_block(
block: OutputContentBlock,
events: &mut Vec<AssistantEvent>,
pending_tool: &mut Option<(String, String, String)>,
streaming_tool_input: bool,
) {
match block {
OutputContentBlock::Text { text } => {
if !text.is_empty() {
events.push(AssistantEvent::TextDelta(text));
}
}
OutputContentBlock::ToolUse { id, name, input } => {
let initial_input = if streaming_tool_input
&& input.is_object()
&& input.as_object().is_some_and(serde_json::Map::is_empty)
{
String::new()
} else {
input.to_string()
};
*pending_tool = Some((id, name, initial_input));
}
}
}
fn response_to_events(response: MessageResponse) -> Vec<AssistantEvent> {
let mut events = Vec::new();
let mut pending_tool = None;
for block in response.content {
push_output_block(block, &mut events, &mut pending_tool, false);
if let Some((id, name, input)) = pending_tool.take() {
events.push(AssistantEvent::ToolUse { id, name, input });
}
}
events.push(AssistantEvent::Usage(response.usage.token_usage()));
events.push(AssistantEvent::MessageStop);
events
}
fn final_assistant_text(summary: &runtime::TurnSummary) -> String {
summary
.assistant_messages
.last()
.map(|message| {
message
.blocks
.iter()
.filter_map(|block| match block {
ContentBlock::Text { text } => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>()
.join("")
})
.unwrap_or_default()
}
#[allow(clippy::needless_pass_by_value)] #[allow(clippy::needless_pass_by_value)]
fn execute_tool_search(input: ToolSearchInput) -> ToolSearchOutput { fn execute_tool_search(input: ToolSearchInput) -> ToolSearchOutput {
let deferred = deferred_tool_specs(); let deferred = deferred_tool_specs();
@@ -2207,7 +2725,7 @@ fn execute_shell_command(
persisted_output_path: None, persisted_output_path: None,
persisted_output_size: None, persisted_output_size: None,
sandbox_status: None, sandbox_status: None,
}); });
} }
let mut process = std::process::Command::new(shell); let mut process = std::process::Command::new(shell);
@@ -2276,7 +2794,7 @@ Command exceeded timeout of {timeout_ms} ms",
persisted_output_path: None, persisted_output_path: None,
persisted_output_size: None, persisted_output_size: None,
sandbox_status: None, sandbox_status: None,
}); });
} }
std::thread::sleep(Duration::from_millis(10)); std::thread::sleep(Duration::from_millis(10));
} }
@@ -2365,6 +2883,7 @@ fn parse_skill_description(contents: &str) -> Option<String> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use std::collections::BTreeSet;
use std::fs; use std::fs;
use std::io::{Read, Write}; use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener}; use std::net::{SocketAddr, TcpListener};
@@ -2373,7 +2892,12 @@ mod tests {
use std::thread; use std::thread;
use std::time::Duration; use std::time::Duration;
use super::{execute_tool, mvp_tool_specs}; use super::{
agent_permission_policy, allowed_tools_for_subagent, execute_agent_with_spawn,
execute_tool, final_assistant_text, mvp_tool_specs, persist_agent_terminal_state,
AgentInput, AgentJob, SubagentToolExecutor,
};
use runtime::{ApiRequest, AssistantEvent, ConversationRuntime, RuntimeError, Session};
use serde_json::json; use serde_json::json;
fn env_lock() -> &'static Mutex<()> { fn env_lock() -> &'static Mutex<()> {
@@ -2765,32 +3289,48 @@ mod tests {
.unwrap_or_else(std::sync::PoisonError::into_inner); .unwrap_or_else(std::sync::PoisonError::into_inner);
let dir = temp_path("agent-store"); let dir = temp_path("agent-store");
std::env::set_var("CLAWD_AGENT_STORE", &dir); std::env::set_var("CLAWD_AGENT_STORE", &dir);
let captured = Arc::new(Mutex::new(None::<AgentJob>));
let captured_for_spawn = Arc::clone(&captured);
let result = execute_tool( let manifest = execute_agent_with_spawn(
"Agent", AgentInput {
&json!({ description: "Audit the branch".to_string(),
"description": "Audit the branch", prompt: "Check tests and outstanding work.".to_string(),
"prompt": "Check tests and outstanding work.", subagent_type: Some("Explore".to_string()),
"subagent_type": "Explore", name: Some("ship-audit".to_string()),
"name": "ship-audit" model: None,
}), },
move |job| {
*captured_for_spawn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(job);
Ok(())
},
) )
.expect("Agent should succeed"); .expect("Agent should succeed");
std::env::remove_var("CLAWD_AGENT_STORE"); std::env::remove_var("CLAWD_AGENT_STORE");
let output: serde_json::Value = serde_json::from_str(&result).expect("valid json"); assert_eq!(manifest.name, "ship-audit");
assert_eq!(output["name"], "ship-audit"); assert_eq!(manifest.subagent_type.as_deref(), Some("Explore"));
assert_eq!(output["subagentType"], "Explore"); assert_eq!(manifest.status, "running");
assert_eq!(output["status"], "queued"); assert!(!manifest.created_at.is_empty());
assert!(output["createdAt"].as_str().is_some()); assert!(manifest.started_at.is_some());
let manifest_file = output["manifestFile"].as_str().expect("manifest file"); assert!(manifest.completed_at.is_none());
let output_file = output["outputFile"].as_str().expect("output file"); let contents = std::fs::read_to_string(&manifest.output_file).expect("agent file exists");
let contents = std::fs::read_to_string(output_file).expect("agent file exists");
let manifest_contents = let manifest_contents =
std::fs::read_to_string(manifest_file).expect("manifest file exists"); std::fs::read_to_string(&manifest.manifest_file).expect("manifest file exists");
assert!(contents.contains("Audit the branch")); assert!(contents.contains("Audit the branch"));
assert!(contents.contains("Check tests and outstanding work.")); assert!(contents.contains("Check tests and outstanding work."));
assert!(manifest_contents.contains("\"subagentType\": \"Explore\"")); assert!(manifest_contents.contains("\"subagentType\": \"Explore\""));
assert!(manifest_contents.contains("\"status\": \"running\""));
let captured_job = captured
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
.expect("spawn job should be captured");
assert_eq!(captured_job.prompt, "Check tests and outstanding work.");
assert!(captured_job.allowed_tools.contains("read_file"));
assert!(!captured_job.allowed_tools.contains("Agent"));
let normalized = execute_tool( let normalized = execute_tool(
"Agent", "Agent",
@@ -2819,6 +3359,195 @@ mod tests {
let _ = std::fs::remove_dir_all(dir); let _ = std::fs::remove_dir_all(dir);
} }
#[test]
fn agent_fake_runner_can_persist_completion_and_failure() {
let _guard = env_lock()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let dir = temp_path("agent-runner");
std::env::set_var("CLAWD_AGENT_STORE", &dir);
let completed = execute_agent_with_spawn(
AgentInput {
description: "Complete the task".to_string(),
prompt: "Do the work".to_string(),
subagent_type: Some("Explore".to_string()),
name: Some("complete-task".to_string()),
model: Some("claude-sonnet-4-6".to_string()),
},
|job| {
persist_agent_terminal_state(
&job.manifest,
"completed",
Some("Finished successfully"),
None,
)
},
)
.expect("completed agent should succeed");
let completed_manifest = std::fs::read_to_string(&completed.manifest_file)
.expect("completed manifest should exist");
let completed_output =
std::fs::read_to_string(&completed.output_file).expect("completed output should exist");
assert!(completed_manifest.contains("\"status\": \"completed\""));
assert!(completed_output.contains("Finished successfully"));
let failed = execute_agent_with_spawn(
AgentInput {
description: "Fail the task".to_string(),
prompt: "Do the failing work".to_string(),
subagent_type: Some("Verification".to_string()),
name: Some("fail-task".to_string()),
model: None,
},
|job| {
persist_agent_terminal_state(
&job.manifest,
"failed",
None,
Some(String::from("simulated failure")),
)
},
)
.expect("failed agent should still spawn");
let failed_manifest =
std::fs::read_to_string(&failed.manifest_file).expect("failed manifest should exist");
let failed_output =
std::fs::read_to_string(&failed.output_file).expect("failed output should exist");
assert!(failed_manifest.contains("\"status\": \"failed\""));
assert!(failed_manifest.contains("simulated failure"));
assert!(failed_output.contains("simulated failure"));
let spawn_error = execute_agent_with_spawn(
AgentInput {
description: "Spawn error task".to_string(),
prompt: "Never starts".to_string(),
subagent_type: None,
name: Some("spawn-error".to_string()),
model: None,
},
|_| Err(String::from("thread creation failed")),
)
.expect_err("spawn errors should surface");
assert!(spawn_error.contains("failed to spawn sub-agent"));
let spawn_error_manifest = std::fs::read_dir(&dir)
.expect("agent dir should exist")
.filter_map(Result::ok)
.map(|entry| entry.path())
.filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("json"))
.find_map(|path| {
let contents = std::fs::read_to_string(&path).ok()?;
contents
.contains("\"name\": \"spawn-error\"")
.then_some(contents)
})
.expect("failed manifest should still be written");
assert!(spawn_error_manifest.contains("\"status\": \"failed\""));
assert!(spawn_error_manifest.contains("thread creation failed"));
std::env::remove_var("CLAWD_AGENT_STORE");
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn agent_tool_subset_mapping_is_expected() {
let general = allowed_tools_for_subagent("general-purpose");
assert!(general.contains("bash"));
assert!(general.contains("write_file"));
assert!(!general.contains("Agent"));
let explore = allowed_tools_for_subagent("Explore");
assert!(explore.contains("read_file"));
assert!(explore.contains("grep_search"));
assert!(!explore.contains("bash"));
let plan = allowed_tools_for_subagent("Plan");
assert!(plan.contains("TodoWrite"));
assert!(plan.contains("StructuredOutput"));
assert!(!plan.contains("Agent"));
let verification = allowed_tools_for_subagent("Verification");
assert!(verification.contains("bash"));
assert!(verification.contains("PowerShell"));
assert!(!verification.contains("write_file"));
}
#[derive(Debug)]
struct MockSubagentApiClient {
calls: usize,
input_path: String,
}
impl runtime::ApiClient for MockSubagentApiClient {
fn stream(&mut self, request: ApiRequest) -> Result<Vec<AssistantEvent>, RuntimeError> {
self.calls += 1;
match self.calls {
1 => {
assert_eq!(request.messages.len(), 1);
Ok(vec![
AssistantEvent::ToolUse {
id: "tool-1".to_string(),
name: "read_file".to_string(),
input: json!({ "path": self.input_path }).to_string(),
},
AssistantEvent::MessageStop,
])
}
2 => {
assert!(request.messages.len() >= 3);
Ok(vec![
AssistantEvent::TextDelta("Scope: completed mock review".to_string()),
AssistantEvent::MessageStop,
])
}
_ => panic!("unexpected mock stream call"),
}
}
}
#[test]
fn subagent_runtime_executes_tool_loop_with_isolated_session() {
let _guard = env_lock()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let path = temp_path("subagent-input.txt");
std::fs::write(&path, "hello from child").expect("write input file");
let mut runtime = ConversationRuntime::new(
Session::new(),
MockSubagentApiClient {
calls: 0,
input_path: path.display().to_string(),
},
SubagentToolExecutor::new(BTreeSet::from([String::from("read_file")])),
agent_permission_policy(),
vec![String::from("system prompt")],
);
let summary = runtime
.run_turn("Inspect the delegated file", None)
.expect("subagent loop should succeed");
assert_eq!(
final_assistant_text(&summary),
"Scope: completed mock review"
);
assert!(runtime
.session()
.messages
.iter()
.flat_map(|message| message.blocks.iter())
.any(|block| matches!(
block,
runtime::ContentBlock::ToolResult { output, .. }
if output.contains("hello from child")
)));
let _ = std::fs::remove_file(path);
}
#[test] #[test]
fn agent_rejects_blank_required_fields() { fn agent_rejects_blank_required_fields() {
let missing_description = execute_tool( let missing_description = execute_tool(