feat(agents): stream provider observations - #159
Conversation
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughProvider-specific observation and token usage parsers were added for five local-agent providers. Runtime adapters now emit observations and return usage. CLI output aggregates and displays usage, with tests covering provider extraction and workflow output. ChangesProvider observability
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Waishnav
changed the title
Greptile SummaryThe PR adds normalized provider activity and token-usage observations across Codex, Claude, OpenCode, Pi, and ACP adapters.
Confidence Score: 5/5The PR appears safe to merge, with no concrete blocking or independently actionable non-blocking defect established. Provider observations flow through the existing workflow persistence path, usage fields are normalized and aggregated consistently with existing workflow views, and no reachable incorrect behavior remains supported by the available repository evidence.
|
| Filename | Overview |
|---|---|
| src/local-agent-provider-observations.ts | Adds provider-independent structural extraction and normalization for tool activity and token usage. |
| src/local-agent-adapters.ts | Connects Claude, OpenCode, ACP, and Pi provider payloads to observation emission and final usage reporting. |
| src/local-agent-runtime.ts | Adds observation deduplication and Codex activity and usage extraction. |
| src/cli-output.ts | Exposes per-call usage and field-wise aggregate workflow-run usage. |
| src/local-agent-provider-observations.test.ts | Covers representative activity and usage payloads for all supported provider families. |
Sequence Diagram
sequenceDiagram
participant Provider
participant Adapter
participant Extractor as Observation Extractor
participant Workflow as Workflow Runtime
participant Store
participant CLI
Provider->>Adapter: events and final response
Adapter->>Extractor: provider payloads
Extractor-->>Adapter: activity and usage observations
Adapter->>Workflow: observations and run result
Workflow->>Store: persist activity and latest usage
Store-->>CLI: call records
CLI->>CLI: sum final call usage
Reviews (1): Last reviewed commit: "feat(agents): wire provider observation ..." | Re-trigger Greptile
|
@coderabbitai full review |
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/local-agent-adapters.ts`:
- Around line 119-125: Update src/local-agent-adapters.ts lines 119-125 in the
Claude adapter to emit one kind: "usage" observation containing the collected
usage after the message loop, and update lines 290-332 in the ACP adapter to
emit the same observation before returning. Add adapter tests verifying
onObservation receives the usage record for both adapters.
- Around line 461-478: Update the final usage assignment in the request flow
around agentEnd and sessionMessages so extractPiUsage(sessionMessages) takes
precedence over extractPiUsage(agentEnd), while retaining the existing usage
fallback. Keep the subsequent usage observation emission unchanged.
In `@src/local-agent-provider-observations.ts`:
- Around line 186-187: Extend the usage-key lists in the normalizer at
src/local-agent-provider-observations.ts:186-187 to read Claude’s
cache_read_input_tokens for cacheRead and cache_creation_input_tokens for
cacheWrite, while preserving existing aliases and fallback behavior. Add or
update coverage in src/local-agent-provider-observations.test.ts:33-40 asserting
these values normalize to cacheReadTokens and cacheWriteTokens.
In `@src/local-agent-runtime.ts`:
- Around line 146-151: The Codex and OpenCode runtime adapters currently emit
observations only after their SDK runs complete. Update
CodexSdkLocalAgentRuntime.run in src/local-agent-runtime.ts:146-151 to subscribe
to the active thread.run event stream and emit observations as events arrive
while retaining final usage emission; likewise update
OpencodeLocalAgentAdapter.run in src/local-agent-adapters.ts:247-251 to consume
waitForOpencodeSession events and emit OpenCode observations during execution.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: af89b973-7325-40b3-930d-9404b67f7ae5
📒 Files selected for processing (7)
package.jsonsrc/cli-output.test.tssrc/cli-output.tssrc/local-agent-adapters.tssrc/local-agent-provider-observations.test.tssrc/local-agent-provider-observations.tssrc/local-agent-runtime.ts
| let usage: LocalAgentRunResult["usage"]; | ||
| const emitObservation = createLocalAgentObservationEmitter(input); | ||
| const items: unknown[] = []; | ||
| for await (const message of messages) { | ||
| items.push(message); | ||
| for (const observation of extractClaudeObservations(message)) emitObservation(observation); | ||
| usage = extractClaudeUsage(message) ?? usage; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Emit collected usage to observation consumers.
Both adapters retain usage for the final result but never send a kind: "usage" observation. onObservation consumers therefore receive tool activity but not token usage.
src/local-agent-adapters.ts#L119-L125: emit the final collected usage after the Claude message loop.src/local-agent-adapters.ts#L290-L332: emit the final collected usage before returning from the ACP run.
Add adapter tests that assert onObservation receives the usage record.
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] Importing child_process exposes a command-execution surface; ensure any command/argument built from input is validated, and prefer execFile/spawn with an argument array over exec.
Context: import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process";
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(detect-child-process-typescript)
📍 Affects 1 file
src/local-agent-adapters.ts#L119-L125(this comment)src/local-agent-adapters.ts#L290-L332
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/local-agent-adapters.ts` around lines 119 - 125, Update
src/local-agent-adapters.ts lines 119-125 in the Claude adapter to emit one
kind: "usage" observation containing the collected usage after the message loop,
and update lines 290-332 in the ACP adapter to emit the same observation before
returning. Add adapter tests verifying onObservation receives the usage record
for both adapters.
| const emitObservation = createLocalAgentObservationEmitter(input); | ||
| let usage: LocalAgentRunResult["usage"]; | ||
| rpc.onEvent((event) => { | ||
| events.push(event); | ||
| for (const observation of extractPiObservations(event)) emitObservation(observation); | ||
| usage = extractPiUsage(event) ?? usage; | ||
| }); | ||
| try { | ||
| const state = await rpc.request({ type: "get_state" }); | ||
| const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null; | ||
| const done = rpc.waitForEvent((event) => asRecord(event)?.type === "agent_end", PI_AGENT_TIMEOUT_MS); | ||
| await rpc.request({ type: "prompt", message: input.prompt }); | ||
| const agentEnd = await done; | ||
| const sessionMessages = await rpc.request({ type: "get_messages" }); | ||
| for (const observation of extractPiObservations(agentEnd)) emitObservation(observation); | ||
| for (const observation of extractPiObservations(sessionMessages)) emitObservation(observation); | ||
| usage = extractPiUsage(agentEnd) ?? extractPiUsage(sessionMessages) ?? usage; | ||
| if (usage) emitObservation({ kind: "usage", usage }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
For @earendil-works/pi-coding-agent version 0.80.3 RPC mode, determine whether get_messages returns the latest cumulative token usage after agent_end.
💡 Result:
In @earendil-works/pi-coding-agent version 0.80.3 RPC mode, the get_messages command returns the list of messages in the current conversation [1][2], but it does not return cumulative token usage [3][4]. To retrieve cumulative token usage in RPC mode, you should use the get_session_stats command [3][5]. The response from get_session_stats includes a tokens object containing input, output, cacheRead, cacheWrite, and total token usage for the session [5][6]. The agent_end event, which is emitted when an agent run completes, is also focused on returning generated messages rather than current session-wide cumulative token metrics [7][5].
Citations:
- 1: https://github.com/earendil-works/pi/blob/v0.80.3/packages/coding-agent/docs/rpc.md
- 2: https://github.com/badlogic/pi-mono/blob/main/packages/coding-agent/docs/rpc.md
- 3: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/src/modes/rpc/rpc-types.ts
- 4: https://github.com/badlogic/pi-mono/blob/main/packages/coding-agent/src/modes/rpc/rpc-client.ts
- 5: https://github.com/badlogic/pi-mono/blob/38f18be4/packages/coding-agent/docs/rpc.md
- 6: https://github.com/badlogic/pi-mono/blob/dd6bea41/packages/coding-agent/src/core/agent-session.ts
- 7: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files matching local-agent-adapters.ts:\n'
fd -a 'local-agent-adapters\.ts$' . || true
file="$(fd 'local-agent-adapters\.ts$' . | head -n1)"
if [ -z "${file:-}" ]; then
printf 'No src/local-agent-adapters.ts found.\n'
exit 0
fi
printf '\nOutline:\n'
ast-grep outline "$file" --view compact | sed -n '1,220p'
printf '\nRelevant lines 430-495:\n'
cat -n "$file" | sed -n '430,495p'
printf '\nSearch for extractPiUsage and Pi usages:\n'
rg -n "extractPiUsage|agent_end|sessionMessages|get_messages|get_session_stats" "$file"Repository: Waishnav/devspace
Length of output: 403
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
for pkg in package.json pnpm-lock.yaml package-lock.json yarn.lock bun.lockb; do
if [ -f "$repo/package.json" ]; then
echo "package.json dependency list:"
node - <<'JS'
const fs = require('fs');
const p = JSON.parse(fs.readFileSync('package.json','utf8'));
for (const section of ['dependencies','devDependencies','peerDependencies','optionalDependencies']) {
console.log(section);
if (p[section]) for (const [k,v] of Object.entries(p[section])) if (k.includes('pi') || k.includes('`@earendil`')) console.log(`${k}: ${v}`);
}
JS
fi
done
printf '\nPackage lock mentions:\n'
for f in pnpm-lock.yaml package-lock.json yarn.lock bun.lockb; do
[ -f "$repo/$f" ] && rg -n "earendil|pi-coding-agent|pi coding agent|`@earendil-works/pi`" "$repo/$f" || true
doneRepository: Waishnav/devspace
Length of output: 14517
🌐 Web query:
site:github.com/earendil-works/pi coding-agent rpc get_messages session messages usage agent_end
💡 Result:
In the context of the coding-agent (part of the earendil-works/pi repository), RPC mode allows for headless interaction with the agent via a JSON-based protocol over stdin/stdout [1]. The terms you mentioned relate to the session lifecycle and state management events: get_messages This is an RPC command used to retrieve the current collection of messages in the active session [2][1]. When called, the agent returns a response containing the full list of AgentMessage objects [2]. agent_end This is a critical lifecycle event emitted when a low-level agent run completes [3][4]. - It signals that the current processing turn (which may include multiple message updates or tool executions) has finished [3]. - The event payload includes a messages field, which contains all messages generated during that specific run, and a willRetry boolean indicating whether an automatic retry is scheduled [3][5]. - It is distinct from agent_settled, which signifies that the agent is fully idle with no further retries or background tasks remaining [3][5]. Session Usage in RPC In RPC mode, session state is managed via events streamed to stdout and commands sent to stdin [1]. - The agent emits lifecycle events such as agent_start, agent_end, turn_start, and turn_end [3][6]. - Developers can integrate with these by subscribing to the stream or using an RpcClient [4][7]. - For state-heavy applications, commands like get_messages, get_entries, and get_tree provide access to the historical and current state of the session [2][8]. If you are building an integration, it is recommended to monitor the agent_end event to determine when it is safe to interact with the session or to process the final results of an agent run [3]. Note that during certain operations like compaction, some incoming commands may be acknowledged but dropped if the session is not in a ready state [9].
Citations:
- 1: https://github.com/earendil-works/pi/blob/v0.80.3/packages/coding-agent/docs/rpc.md
- 2: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/src/modes/rpc/rpc-types.ts
- 3: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
- 4: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/sdk.md
- 5: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/src/core/agent-session.ts
- 6: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/json.md
- 7: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/test/rpc-example.ts
- 8: RPC: expose session entries and tree (
get_entries,get_tree) earendil-works/pi#5810 - 9: RPC prompt during in-flight compaction: ACKed success:true, then silently dropped (0.82.1) earendil-works/pi#7150
🌐 Web query:
@earendil-works pi coding-agent 0.80.3 RPC usage get_message usage
💡 Result:
In the @earendil-works/pi coding-agent (including version 0.80.3), RPC mode allows for headless operation via a JSON-based protocol over stdin/stdout [1]. The get_messages command is used to retrieve all messages in the current conversation [2][3][1]. To use the get_messages command in RPC mode, you send the following JSON object to the agent's standard input (stdin) [2][3]: {"type": "get_messages"} The agent will respond to standard output (stdout) with a JSON object containing the conversation history [2][3]: { "type": "response", "command": "get_messages", "success": true, "data": { "messages": [...] } } The messages returned are in the form of AgentMessage objects [2][3]. RPC mode is intended for embedding the agent in other applications, IDEs, or custom user interfaces [3][1]. For Node.js/TypeScript integrations, the project documentation recommends using the AgentSession SDK directly instead of spawning a subprocess [1].
Citations:
- 1: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
- 2: https://github.com/earendil-works/pi/blob/v0.80.3/packages/coding-agent/docs/rpc.md
- 3: https://github.com/earendil-works/pi/blob/8b97e75c6b149fdd4dec95fe3321d1e94fd5c1d4/packages/coding-agent/docs/rpc.md
🌐 Web query:
github.com/earendil-works/pi packages/coding-agent src modes rpc rpc-types.ts AgentMessage usage 0.80.3
💡 Result:
In the earendil-works/pi repository (specifically version 0.80.3), AgentMessage is a core type used to define the conversation history and message abstraction for the agent [1][2][3]. It serves as a union of standard LLM message types (user, assistant, toolResult) and custom application-specific message types, allowing for extensibility while maintaining type safety [1][2]. Within the coding-agent package's RPC mode (packages/coding-agent/src/modes/rpc/rpc-types.ts), AgentMessage is used to structure data in responses, particularly when retrieving message history [4]. For instance, a "get_messages" response includes an array of AgentMessage objects [4]. Key points regarding its usage: 1. Definition: AgentMessage is defined in @earendil-works/pi-agent-core (packages/agent/src/types.ts) [1][3]. It allows for declaration merging, enabling users to add custom message types beyond the standard LLM formats [2][3]. 2. RPC Context: In the RPC mode, which facilitates headless operation via JSON-RPC over stdin/stdout, AgentMessage is the primary type used for message-related commands [4][5]. 3. Transformation: Because LLMs natively only understand standard message roles, the AgentMessage objects must be transformed into LLM-compatible formats using a convertToLlm function, which filters out custom or UI-only messages [1][2][6]. 4. Developer Implementation: Developers using the RPC mode can utilize the provided RpcClient or implement their own client to consume these messages [5][7]. The test file packages/coding-agent/test/rpc-example.ts provides an interactive example of how to handle agent events and messages [7]. For further implementation details, you can refer to the official documentation in the repository at packages/coding-agent/docs/rpc.md and the source definition in packages/agent/src/types.ts [5][1].
Citations:
- 1: https://github.com/earendil-works/pi/blob/main/packages/agent/src/types.ts
- 2: https://github.com/earendil-works/pi/blob/main/packages/agent/README.md
- 3: https://www.npmjs.com/package/@earendil-works/pi-agent-core
- 4: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/src/modes/rpc/rpc-types.ts
- 5: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
- 6: https://github.com/earendil-works/pi/blob/main/packages/agent/src/agent-loop.ts
- 7: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/test/rpc-example.ts
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\nResolved package cache/source snippets if available:\n'
for f in \
node_modules/@earendil-works/pi-coding-agent/lib/modes/rpc/rpc-types.js \
node_modules/@earendil-works/pi-coding-agent/dist/modes/rpc/rpc-types.js \
node_modules/@earendil-works/pi-coding-agent/src/modes/rpc/rpc-types.ts; do
if [ -f "$f" ]; then
echo "--- $f"
rg -n "AgentMessage|usage|AgentRun|agent_end|get_messages|completion" "$f" -C 3 || true
fi
done
printf '\nDownloaded npm tarball schema/source snippets:\n'
tmp="$(mktemp -d)"
curl -fsSL https://registry.npmjs.org/@earendil-works/pi-coding-agent/-/pi-coding-agent-0.80.3.tgz -o "$tmp/pkg.tgz"
tar -tzf "$tmp/pkg.tgz" | rg 'src/modes/rpc/rpc-types|src/modes/rpc/rpc-client|src/.*agent' | head -n 80
echo "--- rpc-types"
tar -xOzf "$tmp/pkg.tgz" "$tmp/pkg.tgz" 2>/dev/null >/tmp/pkg.tgz >/tmp/.err || true
tar -xOzf "$tmp/pkg.tgz" `tar -tzf "$tmp/pkg.tgz" 2>/dev/null | rg 'src/modes/rpc/rpc-types' | head -n1` 2>/dev/null | rg -n "AgentMessage|usage|agent" -C 3 || true
echo "--- likely agent usage files"
for s in `tar -tzf "$tmp/pkg.tgz" 2>/dev/null | rg 'src|lib|dist' | grep -E '(agent|session|usage|token)' | head -n 20`; do
echo "--- $s"
tar -xOzf "$tmp/pkg.tgz" "$s" 2>/dev/null | rg -n "usage|tokens|AgentMessage|agent_end|get_messages" -C 2 || true
doneRepository: Waishnav/devspace
Length of output: 257
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
printf 'Relevant lines src/local-agent-adapters.ts:\n'
cat -n src/local-agent-adapters.ts | sed -n '1,120p;380,520p'
printf '\nDefinitions/usages of extractPiUsage and Pi adapter path:\n'
rg -n "extractPiUsage|extractPiObservations|extract.*Usage|createLocalAgentObservationEmitter|provider: 'pi'|`@earendil`|pi-coding-agent|prompt: input.prompt" src/local-agent-adapters.ts src packages tests -g '*.ts' -g '*.tsx' -C 2 || true
printf '\nBehavioral probe of nullish precedence order:\n'
node - <<'JS'
function usedValue(agentEnd, sessionMessages, usage) {
return agentEnd ?? sessionMessages ?? usage;
}
function preferredValue(sessionMessages, agentEnd, usage) {
return sessionMessages ?? agentEnd ?? usage;
}
const cases = [
{agentEnd: {tokens: 1}, sessionMessages: {tokens: 2}, usage: null},
{agentEnd: null, sessionMessages: {tokens: 2}, usage: {tokens: 0}},
{agentEnd: {tokens: 1}, sessionMessages: {tokens: 2}, usage: {tokens: 3}},
];
for (const c of cases) console.log(JSON.stringify({input: c, first: usedValue(c.agentEnd, c.sessionMessages, c.usage), later: preferredValue(c.sessionMessages, c.agentEnd, c.usage)}));
JSRepository: Waishnav/devspace
Length of output: 33006
🌐 Web query:
"PiRpcLocalAgentAdapter" "extractPiUsage"
💡 Result:
The identifier PiRpcLocalAgentAdapter does not appear to be a standard class or component in the official Pi coding agent repository, but extractPiUsage is a utility function used in community-developed tools, CLI wrappers, and adapters to aggregate token usage and cost data from Pi session logs [1]. In the context of Pi's headless operation (often via --mode rpc) [2][3], session activity is recorded in JSONL files [4][3]. Community projects and custom adapters use functions like extractPiUsage to parse these logs and generate summaries [1]. Key characteristics of extractPiUsage implementations include: 1. Data Source: It processes an array of JSON records (typically parsed from a Pi session.jsonl file) [1][5]. 2. Functionality: It searches for the latest message containing usage statistics (e.g., input tokens, output tokens, cache reads/writes, and costs) [1]. 3. Output: It returns a summary object (often called UsageSummary) that includes totals for tokens and costs [1]. 4. Integration: It is commonly used in extensions or CLI tools to provide usage dashboards or cost tracking [4][6][7]. If you are encountering this in a specific codebase, it is likely part of a custom adapter (such as a Paperclip adapter or a headless CLI wrapper) designed to interface with the Pi coding agent's RPC mode output [8][1][9]. These tools parse the streaming JSONL output or historical session files to reconstruct token consumption for billing or monitoring purposes [4][6][10]. For further implementation details specific to your environment, check the source code of the adapter or extension where the function is defined, as these are often project-specific utilities [1][11].
Citations:
- 1: RobertTLange/headless-cli@cb6b8ee
- 2: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
- 3: https://deepwiki.com/agentic-dev-io/pi-agent/7-rpc-mode-and-headless-integration
- 4: https://cdn.jsdelivr.net/npm/pi-extensions@0.1.58/usage-extension/README.md
- 5: https://github.com/seanpedrick-case/doc_redaction/blob/main/agent-redact/pi/pi_session_usage.py
- 6: https://github.com/pi-vault/pi-usage
- 7: https://github.com/ZaganJade/pi-extension/blob/main/usage/src/index.ts
- 8: https://github.com/paperclipai/paperclip/blob/f12bb27b/packages/adapters/pi-local/src/server/parse.ts
- 9: https://github.com/paperclipai/paperclip/blob/f12bb27b/packages/adapters/pi-local/src/index.ts
- 10: https://knightli.com/en/2026/07/28/pi-coding-agent-rpc-json-protocol-provider-custom-web-ui/
- 11: https://github.com/ZaganJade/pi-extension/blob/main/usage/src/aggregate.ts
🌐 Web query:
@earendil-works pi-coding-agent 0.80.3 source usage UpdateEvent
💡 Result:
In the earendil-works/pi coding-agent version 0.80.3, UpdateEvent does not exist as a standalone top-level event type in the agent's RPC event stream [1]. Instead, the agent utilizes specific streaming event types for updates during operation, emitted as JSON lines to stdout [1]. The primary event for streaming updates is message_update, which provides deltas (text, thinking, or tool calls) as the model generates content [1]. Additionally, tool_execution_update is used specifically to report progress and partial output during tool execution [1][2][3]. Key streaming event types for agent operations include [1]: - message_update: Contains partial assistant messages (text chunks, thinking deltas, or tool call arguments). - tool_execution_update: Used for streaming output from tool executions. - queue_update: Signifies changes to pending steering or follow-up queues. In the internal implementation, these events are processed by an agent event sink [2][3]. Developers integrating with the agent or building extensions typically listen for these specific event types rather than a generic UpdateEvent [1][4]. If you are working with RPC integrations or custom extensions, refer to the documentation in packages/coding-agent/docs/rpc.md within the repository for the full schema of these events [1].
Citations:
- 1: https://github.com/earendil-works/pi/blob/v0.80.3/packages/coding-agent/docs/rpc.md
- 2: https://github.com/earendil-works/pi/blob/main/packages/agent/src/agent-loop.ts
- 3: https://github.com/badlogic/pi-mono/blob/f3a2c9d0/packages/agent/src/agent-loop.ts
- 4: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/sdk.md
🌐 Web query:
@earendil-works pi coding-agent get_session_stats returns usage tokens
💡 Result:
Yes, the get_session_stats function in the @earendil-works/pi coding-agent returns comprehensive usage token information [1][2]. When called, this RPC method returns a payload containing detailed session statistics, including a nested tokens object that breaks down usage by category [1][3]: tokens: { input: number, output: number, cacheRead: number, cacheWrite: number, total: number } These counts represent cumulative usage across the full session, including assistant messages, usage reported by tools, and overhead from compaction or branch-summary generation [1][4]. In addition to these cumulative token totals, the response also includes a contextUsage field, which provides the actual current context-window estimate used for compaction and display [1][5]. This field was explicitly added to allow RPC clients to track current window usage separately from the monotonically increasing cumulative token counts [5]. If no model or context window is available, the contextUsage field may be omitted, or its properties may be null (for example, immediately following compaction) [1][6].
Citations:
- 1: https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc.md
- 2: https://github.com/badlogic/pi-mono/blob/main/packages/coding-agent/docs/rpc.md
- 3: https://docs.bastani.ai/rpc
- 4: https://github.com/earendil-works/pi/blob/8b97e75c6b149fdd4dec95fe3321d1e94fd5c1d4/packages/coding-agent/docs/rpc.md
- 5: Expose contextUsage in get_session_stats RPC response earendil-works/pi#2550
- 6: https://github.com/earendil-works/pi/blob/v0.82.0/packages/coding-agent/docs/rpc.md
Use the later Pi usage payload for the final result.
get_messages is fetched after agent_end, so this precedence can keep an older usage record when both payloads contain usage. Reverse the operands before emitting usage.
Proposed fix
- usage = extractPiUsage(agentEnd) ?? extractPiUsage(sessionMessages) ?? usage;
+ usage = extractPiUsage(sessionMessages) ?? extractPiUsage(agentEnd) ?? usage;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const emitObservation = createLocalAgentObservationEmitter(input); | |
| let usage: LocalAgentRunResult["usage"]; | |
| rpc.onEvent((event) => { | |
| events.push(event); | |
| for (const observation of extractPiObservations(event)) emitObservation(observation); | |
| usage = extractPiUsage(event) ?? usage; | |
| }); | |
| try { | |
| const state = await rpc.request({ type: "get_state" }); | |
| const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null; | |
| const done = rpc.waitForEvent((event) => asRecord(event)?.type === "agent_end", PI_AGENT_TIMEOUT_MS); | |
| await rpc.request({ type: "prompt", message: input.prompt }); | |
| const agentEnd = await done; | |
| const sessionMessages = await rpc.request({ type: "get_messages" }); | |
| for (const observation of extractPiObservations(agentEnd)) emitObservation(observation); | |
| for (const observation of extractPiObservations(sessionMessages)) emitObservation(observation); | |
| usage = extractPiUsage(agentEnd) ?? extractPiUsage(sessionMessages) ?? usage; | |
| if (usage) emitObservation({ kind: "usage", usage }); | |
| const emitObservation = createLocalAgentObservationEmitter(input); | |
| let usage: LocalAgentRunResult["usage"]; | |
| rpc.onEvent((event) => { | |
| events.push(event); | |
| for (const observation of extractPiObservations(event)) emitObservation(observation); | |
| usage = extractPiUsage(event) ?? usage; | |
| }); | |
| try { | |
| const state = await rpc.request({ type: "get_state" }); | |
| const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null; | |
| const done = rpc.waitForEvent((event) => asRecord(event)?.type === "agent_end", PI_AGENT_TIMEOUT_MS); | |
| await rpc.request({ type: "prompt", message: input.prompt }); | |
| const agentEnd = await done; | |
| const sessionMessages = await rpc.request({ type: "get_messages" }); | |
| for (const observation of extractPiObservations(agentEnd)) emitObservation(observation); | |
| for (const observation of extractPiObservations(sessionMessages)) emitObservation(observation); | |
| usage = extractPiUsage(sessionMessages) ?? extractPiUsage(agentEnd) ?? usage; | |
| if (usage) emitObservation({ kind: "usage", usage }); |
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] Importing child_process exposes a command-execution surface; ensure any command/argument built from input is validated, and prefer execFile/spawn with an argument array over exec.
Context: import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process";
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(detect-child-process-typescript)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/local-agent-adapters.ts` around lines 461 - 478, Update the final usage
assignment in the request flow around agentEnd and sessionMessages so
extractPiUsage(sessionMessages) takes precedence over extractPiUsage(agentEnd),
while retaining the existing usage fallback. Keep the subsequent usage
observation emission unchanged.
| const cacheRead = numberValue(record, ["cacheRead", "cache_read", "cacheReadTokens", "cache_read_tokens"]) ?? numberValue(cache, ["read", "readTokens", "read_tokens"]); | ||
| const cacheWrite = numberValue(record, ["cacheWrite", "cache_write", "cacheWriteTokens", "cache_write_tokens"]) ?? numberValue(cache, ["write", "writeTokens", "write_tokens"]); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
For @anthropic-ai/claude-agent-sdk version 0.3.200, verify the exact cache-token field names forwarded in result usage payloads.
💡 Result:
For @anthropic-ai/claude-agent-sdk@0.3.200, the ResultMessage.usage payload forwards these snake_case cache fields:
cache_creation_input_tokenscache_read_input_tokens
They accompany input_tokens and output_tokens; they are not camel-cased in usage. CamelCase names such as cacheReadInputTokens and cacheCreationInputTokens apply to the per-model model_usage structure instead. (code.claude.com)
Version 0.3.200 was published with no reported change to these usage field names. (sourceforge.net)
Citations:
- 1: https://code.claude.com/docs/en/agent-sdk/cost-tracking?utm_source=openai
- 2: https://sourceforge.net/projects/claude-agent-sdk.mirror/files/v0.3.200/?utm_source=openai
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== candidate files =="
git ls-files | rg '(^|/)local-agent-provider-observations\.(ts|test\.ts)$|package\.json$|pnpm-lock\.yaml$|package-lock\.json$|yarn\.lock$' || true
echo
echo "== observations target =="
if [ -f src/local-agent-provider-observations.ts ]; then
nl -ba src/local-agent-provider-observations.ts | sed -n '160,205p'
fi
echo
echo "== observations tests around Claude fixture =="
if [ -f src/local-agent-provider-observations.test.ts ]; then
nl -ba src/local-agent-provider-observations.test.ts | sed -n '1,80p'
fi
echo
echo "== package versions/locks for anthropic SDK =="
for f in package.json pnpm-lock.yaml package-lock.json yarn.lock; do
if [ -f "$f" ]; then
echo "--- $f ---"
rg -n "`@anthropic-ai/claude-agent-sdk`|claude-agent-sdk|version" "$f" | head -60 || true
fi
doneRepository: Waishnav/devspace
Length of output: 363
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== observations target =="
if [ -f src/local-agent-provider-observations.ts ]; then
awk '{print NR": "$0}' src/local-agent-provider-observations.ts | sed -n '160,205p'
fi
echo
echo "== observations tests around Claude fixture =="
if [ -f src/local-agent-provider-observations.test.ts ]; then
awk '{print NR": "$0}' src/local-agent-provider-observations.test.ts | sed -n '1,80p'
fi
echo
echo "== package versions/locks for anthropic SDK =="
for f in package.json pnpm-lock.yaml package-lock.json yarn.lock; do
if [ -f "$f" ]; then
echo "--- $f ---"
grep -nE "`@anthropic-ai/claude-agent-sdk`|claude-agent-sdk|version" "$f" | head -60 || true
fi
doneRepository: Waishnav/devspace
Length of output: 9769
Preserve Claude cache-token usage.
Claude 0.3.200 result usage uses result.usage.cache_read_input_tokens and result.usage.cache_creation_input_tokens, but normalizer keys at src/local-agent-provider-observations.ts#L186-L187 stop before reading those fields. Add both fields to Claude result usage usage reading, and cover the normalized cacheReadTokens/cacheWriteTokens values in src/local-agent-provider-observations.test.ts#L33-L40.
Proposed fix
- const cacheRead = numberValue(record, ["cacheRead", "cache_read", "cacheReadTokens", "cache_read_tokens"]) ?? numberValue(cache, ["read", "readTokens", "read_tokens"]);
+ const cacheRead = numberValue(record, ["cacheRead", "cache_read", "cacheReadTokens", "cache_read_tokens", "cache_read_input_tokens"]) ?? numberValue(cache, ["read", "readTokens", "read_tokens"]);- const cacheWrite = numberValue(record, ["cacheWrite", "cache_write", "cacheWriteTokens", "cache_write_tokens"]) ?? numberValue(cache, ["write", "writeTokens", "write_tokens"]);
+ const cacheWrite = numberValue(record, ["cacheWrite", "cache_write", "cacheWriteTokens", "cache_write_tokens", "cache_creation_input_tokens"]) ?? numberValue(cache, ["write", "writeTokens", "write_tokens"]);📍 Affects 2 files
src/local-agent-provider-observations.ts#L186-L187(this comment)src/local-agent-provider-observations.test.ts#L33-L40
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/local-agent-provider-observations.ts` around lines 186 - 187, Extend the
usage-key lists in the normalizer at
src/local-agent-provider-observations.ts:186-187 to read Claude’s
cache_read_input_tokens for cacheRead and cache_creation_input_tokens for
cacheWrite, while preserving existing aliases and fallback behavior. Add or
update coverage in src/local-agent-provider-observations.test.ts:33-40 asserting
these values normalize to cacheReadTokens and cacheWriteTokens.
| const emitObservation = createLocalAgentObservationEmitter(input); | ||
| for (const item of turn.items) { | ||
| for (const observation of extractCodexObservations(item)) emitObservation(observation); | ||
| } | ||
| const usage = extractCodexUsage(turn); | ||
| if (usage) emitObservation({ kind: "usage", usage }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
For @openai/codex-sdkversion 0.142.5 and@opencode-ai/sdk version 1.17.13, identify the documented APIs for receiving turn or session events while execution is active.
💡 Result:
For the specified SDK versions, turn session events are handled through the following documented APIs: @openai/codex-sdk (version 0.142.5) To receive turn session events while execution is active, use the runStreamed method [1][2]. This method returns an async generator that yields structured events as they occur, allowing you to react to intermediate progress such as tool calls, streaming responses, and file changes [1][2]. Example usage: const { events } = await thread.runStreamed("Your prompt here"); for await (const event of events) { switch (event.type) { case "item.completed": console.log("Item completed", event.item); break; case "turn.completed": console.log("Turn completed", event.usage); break; } } The SDK also supports various event types such as item.started, item.updated, item.completed, turn.completed, and turn.failed, which can be handled within the loop [3]. @opencode-ai/sdk (version 1.17.13) To receive real-time events while a session is active, use the client.event.subscribe method [4][5]. This returns a server-sent events (SSE) stream [4]. Example usage: const events = await client.event.subscribe; for await (const event of events.stream) { if (event.type === "session.status") { console.log(event.properties.sessionID, event.properties.status.type); } } Documented session events include session.created, session.updated, session.deleted, session.diff, and session.status (which indicates whether the session is idle, busy, or in a retry state) [5]. Additionally, V2 session lifecycle events (session.next.*) are available for newer session engines [5].
Citations:
- 1: https://github.com/openai/codex/blob/main/sdk/typescript/README.md
- 2: https://github.com/openai/codex/tree/main/sdk/typescript
- 3: https://mintlify.wiki/openai/codex/sdk/usage
- 4: https://opencode.ai/docs/sdk/
- 5: https://github.com/uplift-labs/opencode-dev-docs/blob/main/03-events-reference.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== package versions =="
for f in package.json package-lock.json pnpm-lock.yaml yarn.lock; do
[ -f "$f" ] && { echo "--- $f"; grep -E '`@openai/codex-sdk`|`@opencode-ai/sdk`' "$f" | head -20 || true; }
done
echo "== locate relevant files =="
fd -a 'local-agent-runtime\.ts|local-agent-adapters\.ts|package\.json' . | sed 's#^\./##'
echo "== inspect relevant ranges =="
for f in src/local-agent-runtime.ts src/local-agent-adapters.ts; do
if [ -f "$f" ]; then
echo "--- $f line count"
wc -l "$f"
echo "--- $f lines 1-340"
sed -n '1,340p' "$f" | nl -ba | sed -n '1,340p'
fi
done
echo "== search SDK usage patterns =="
rg -n "runStreamed|waitForOpencodeSession|createLocalAgentObservationEmitter|extractCodexObservations|extractOpenCodeObservations|event\.subscribe|thread\.run\(" src package.json package-lock.json pnpm-lock.yaml yarn.lock 2>/dev/null || trueRepository: Waishnav/devspace
Length of output: 963
🌐 Web query:
@openai/codex-sdk 0.142.5 runStreamed events item.completed turn.completed tool call
💡 Result:
In the @openai/codex-sdk, the runStreamed method returns an async generator of structured events, allowing you to process intermediate progress such as tool calls and streaming responses [1][2]. The events item.completed and turn.completed are key types emitted by this generator [3][4]. 1. item.completed: This event signals that a specific thread item (such as an agent message, reasoning output, or command execution) has reached a terminal state (success or failure) [3][5][4]. The event payload includes the item object, which contains details about the completed work, such as exit codes for command executions or the text content of messages [5]. 2. turn.completed: This event is emitted when an entire interaction turn is finished, typically immediately after the assistant has completed its response [3][4]. It contains usage data, including input, output, and reasoning token counts [5]. Example usage: const { events } = await thread.runStreamed("Your prompt here"); for await (const event of events) { switch (event.type) { case "item.completed": console.log("Item completed:", event.item); break; case "turn.completed": console.log("Turn completed. Usage:", event.usage); break; } } This streaming approach is preferred over the standard run method when you need to react to real-time progress rather than waiting for the entire turn to buffer [1][2].
Citations:
- 1: https://github.com/openai/codex/tree/main/sdk/typescript
- 2: https://github.com/openai/codex/blob/main/sdk/typescript/README.md
- 3: https://github.com/openai/codex/blob/35aaa5d9/sdk/typescript/src/events.ts
- 4: https://github.com/openai/codex/blob/53b50197/sdk/typescript/src/events.ts
- 5: https://github.com/openai/codex/blob/fbe65995/sdk/typescript/samples/basic_streaming.ts
🌐 Web query:
@opencode-ai/sdk 1.17.13 client.event.subscribe SSE session.status
💡 Result:
In the @opencode-ai/sdk version 1.17.13, you can subscribe to real-time events, including session status updates, using the client.event.subscribe method [1][2][3]. This method returns an async iterable stream [4]. You can consume the stream to listen for specific event types, such as session.status [1][3]. Example implementation: // Listen to real-time events const events = await client.event.subscribe; for await (const event of events.stream) { if (event.type === "session.status") { console.log("Session ID:", event.properties.sessionID); console.log("Status:", event.properties.status); } } The session.status event properties contain a sessionID and a status object [5][6][7]. The status can be one of the following types [5][6][8]: - idle: The session is currently inactive [5][8]. - busy: The session is actively processing [5][8]. - retry: The session has encountered an error and is in a retry state, including details like the attempt count, error message, and the time of the next attempt [5][6][7]. Note that while event.subscribe handles real-time streams [1][2], there is also a REST endpoint GET /session/status available on the server that returns the current status of all sessions as a static snapshot [9].
Citations:
- 1: https://open-code.ai/en/docs/sdk
- 2: https://dev.opencode.ai/docs/sdk/
- 3: https://opencode.ai/docs/sdk.md
- 4: https://opencode.ai/v2/docs/build/client
- 5: https://github.com/sst/opencode/blob/dev/packages/sdk/js/src/gen/types.gen.ts
- 6: https://github.com/sst/opencode/blob/5d2dc888/packages/sdk/js/src/v2/gen/types.gen.ts
- 7: https://github.com/sst/opencode/blob/69a80663/packages/sdk/js/src/v2/gen/types.gen.ts
- 8: https://github.com/sst/opencode/blob/47f33329/packages/opencode/src/session/status.ts
- 9: https://opencode.ai/docs/server/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== inspect relevant ranges =="
sed -n '1,340p' src/local-agent-runtime.ts 2>/dev/null | awk '{printf "%6d %s\n", NR, $0}'
echo "---"
sed -n '1,360p' src/local-agent-adapters.ts 2>/dev/null | awk '{printf "%6d %s\n", NR, $0}'
echo "== search SDK usage patterns =="
rg -n "runStreamed|waitForOpencodeSession|createLocalAgentObservationEmitter|extractCodexObservations|extractOpenCodeObservations|thread\.run\(|client\.event\.subscribe|event\.subscribe" src package.json node_modules 2>/dev/null || true
echo "== exact TypeScript snippets around relevant symbols =="
python3 - <<'PY'
from pathlib import Path
for path in [Path("src/local-agent-runtime.ts"), Path("src/local-agent-adapters.ts")]:
text = path.read_text()
for token in ["extractCodexObservations", "extractOpenCodeObservations", "waitForOpencodeSession", "thread.run(", "runStreamed", "createLocalAgentObservationEmitter", "client.event.subscribe", "event.subscribe"]:
idx = text.find(token)
print(f"\n-- {path} find({token!r})={idx}")
if idx >= 0:
line = text.count("\n", 0, idx) + 1
start = max(0, text.rfind("\n", 0, idx - 500))
end = text.find("\n", idx + 2000)
if end == -1: end = len(text)
print(f"around line {line}: {text[start:end].replace(chr(10), f\"\\n[{path.name}:{line}]\").split(chr(10))[:8]}")
PYRepository: Waishnav/devspace
Length of output: 25770
Emit Codex and OpenCode observations from active streams.
CodexSdkLocalAgentRuntime.run emits Codex observations after thread.run(...) completes, and OpencodeLocalAgentAdapter.run emits OpenCode observations after waitForOpencodeSession(...) completes. Add the active SDK event streams so progress such as tool status updates is not invisible until the run finishes.
📍 Affects 2 files
src/local-agent-runtime.ts#L146-L151(this comment)src/local-agent-adapters.ts#L247-L251
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/local-agent-runtime.ts` around lines 146 - 151, The Codex and OpenCode
runtime adapters currently emit observations only after their SDK runs complete.
Update CodexSdkLocalAgentRuntime.run in src/local-agent-runtime.ts:146-151 to
subscribe to the active thread.run event stream and emit observations as events
arrive while retaining final usage emission; likewise update
OpencodeLocalAgentAdapter.run in src/local-agent-adapters.ts:247-251 to consume
waitForOpencodeSession events and emit OpenCode observations during execution.
Source: Coding guidelines
|
Closing this stack in favor of the alternative provider-observability implementation. Review found several unresolved provider-contract and lifecycle issues, including non-live Codex/OpenCode observation paths, incomplete/incorrect provider usage mappings, and assumptions in generic normalizers that did not consistently match the real SDK contracts. The modular normalization approach is useful, but this implementation will not be carried forward. |
1 participant
Agent providers expose different event and token-usage shapes, so workflow observability otherwise stops at the adapter boundary. Structural normalizers now cover Codex, Claude, OpenCode, Pi, and ACP-backed Cursor/Copilot events, while each adapter forwards best-effort tool activity and usage into the workflow journal and CLI output. Values are only emitted when the provider supplies them; unsupported or disabled providers do not produce fabricated observations. This is the top layer of the stack.
Summary by CodeRabbit
New Features
Tests