feat(workflow): observe Codex and Claude agent runs - #153
Conversation
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughLocal-agent runtimes and adapters now report sessions, activities, and token usage through optional observers. Workflow execution persists these updates, associates them with provider call indexes, and flushes pending usage when runs complete. ChangesLocal-agent observability
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant WorkflowWorker
participant LocalAgentProvider
participant LocalAgentRuntime
participant WorkflowAgentObserver
participant WorkflowStore
WorkflowWorker->>WorkflowAgentObserver: create observer(runId, callIndex)
WorkflowWorker->>LocalAgentProvider: run provider with observer
LocalAgentProvider->>LocalAgentRuntime: run(input, observer)
LocalAgentRuntime->>WorkflowAgentObserver: session, activity, and usage callbacks
WorkflowAgentObserver->>WorkflowStore: persist workflow updates
WorkflowWorker->>WorkflowAgentObserver: close()
WorkflowAgentObserver->>WorkflowStore: flush pending usage
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 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 live provider observability for workflow agent calls, including sessions, token usage, and tool activity.
Confidence Score: 4/5The usage accounting defect should be fixed before merging because schema retries currently under-report provider token consumption. Schema enforcement can invoke the provider multiple times under one call index, while each newly created observer replaces the stored usage with its own final snapshot, dropping tokens consumed by earlier attempts. Files Needing Attention: src/workflow-worker.ts, src/workflow-agent-observer.ts
|
| Filename | Overview |
|---|---|
| src/local-agent-runtime.ts | Adds Codex streamed-event collection and normalized session, activity, and token-usage reporting. |
| src/local-agent-adapters.ts | Propagates observers through local-agent adapters and extracts Claude session, activity, and usage events. |
| src/workflow-agent-observer.ts | Introduces throttled persistence for workflow-agent observations, but usage snapshots replace rather than aggregate schema-retry attempts. |
| src/workflow-worker.ts | Creates an observer around every provider invocation, causing retry attempts sharing a call index to overwrite prior usage. |
| src/workflow-api.ts | Adds the stable workflow call index to provider-run inputs. |
| src/local-agent-runtime.test.ts | Covers the normalized Codex streaming happy path for sessions, activities, final responses, and usage. |
| src/workflow-store.test.ts | Covers persistence of observed sessions, activity, and final usage snapshots. |
Sequence Diagram
sequenceDiagram
participant W as Workflow API
participant P as Provider runner
participant O as Workflow observer
participant S as Workflow store
W->>P: Run agent call (callIndex)
P->>O: Session/activity/usage callbacks
O->>S: Persist observations
alt Schema validation fails
W->>P: Retry with same callIndex
P->>O: Create new observer
O->>S: Replace usage snapshot
end
P-->>W: Final provider result
Reviews (1): Last reviewed commit: "feat(workflow): persist live provider ob..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
63e67b8 to
84fc2ee
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
|
@coderabbitai full review |
✅ Action performedFull review finished. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 188-198: Update the token aggregate in the usage return block to
include cacheReadInputTokens and cacheCreationInputTokens alongside inputTokens
and outputTokens when calculating totalTokens. Preserve the existing
non-negative normalization and undefined handling in the surrounding token
fields.
- Around line 127-128: Update the Claude record-processing flow around
claudeUsage and observer.onUsage so usage is parsed for every received record,
emitting a partial snapshot for non-result records and retaining final for the
result record. Add coverage for an intermediate usage record verifying the
observer receives the partial update.
🪄 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: 49413a83-95bf-4d43-9a58-8aa4c182193c
📒 Files selected for processing (7)
src/local-agent-adapters.tssrc/local-agent-runtime.test.tssrc/local-agent-runtime.tssrc/workflow-agent-observer.tssrc/workflow-api.tssrc/workflow-store.test.tssrc/workflow-worker.ts
| const usage = claudeUsage(record.usage, "final"); | ||
| if (usage) observer?.onUsage?.(usage); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Emit partial Claude usage updates.
Lines 127-128 call onUsage only for a result record. Line 119 skips this block for every non-result record. The observer therefore never receives a state: "partial" snapshot.
Parse usage on each received record. Emit partial for non-result records. Keep the result record as final. Add a test with an intermediate usage record.
Proposed fix
notifyClaudeActivity(record, observer);
+ const usage = claudeUsage(record.usage, record.type === "result" ? "final" : "partial");
+ if (usage) observer?.onUsage?.(usage);
if (record.type !== "result") continue;
const resultError = claudeResultError(record);
if (resultError) throw new Error(resultError);
@@
- const usage = claudeUsage(record.usage, "final");
- if (usage) observer?.onUsage?.(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 usage = claudeUsage(record.usage, "final"); | |
| if (usage) observer?.onUsage?.(usage); | |
| notifyClaudeActivity(record, observer); | |
| const usage = claudeUsage(record.usage, record.type === "result" ? "final" : "partial"); | |
| if (usage) observer?.onUsage?.(usage); | |
| if (record.type !== "result") continue; |
🧰 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 127 - 128, Update the Claude
record-processing flow around claudeUsage and observer.onUsage so usage is
parsed for every received record, emitting a partial snapshot for non-result
records and retaining final for the result record. Add coverage for an
intermediate usage record verifying the observer receives the partial update.
| const inputTokens = nonNegativeInteger(usage.input_tokens); | ||
| const outputTokens = nonNegativeInteger(usage.output_tokens); | ||
| if (inputTokens === undefined && outputTokens === undefined) return undefined; | ||
| return { | ||
| inputTokens, | ||
| cachedInputTokens: nonNegativeInteger(usage.cache_read_input_tokens), | ||
| cacheCreationInputTokens: nonNegativeInteger(usage.cache_creation_input_tokens), | ||
| outputTokens, | ||
| totalTokens: (inputTokens ?? 0) + (outputTokens ?? 0), | ||
| state, | ||
| }; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files matching local-agent-adapters.ts:\n'
fd -a 'local-agent-adapters\.ts$' . || true
printf '\nRelevant file excerpt:\n'
file="$(fd 'local-agent-adapters\.ts$' . | head -n 1 || true)"
if [ -n "${file:-}" ]; then
wc -l "$file"
sed -n '150,220p' "$file" | nl -ba -v150
fi
printf '\nSearch usage of function/totalTokens/cached tokens:\n'
rg -n "totalTokens|cache_read_input_tokens|cache_creation_input_tokens|inputTokens|cachedInputTokens|cacheCreationInputTokens|nonNegativeInteger" .Repository: Waishnav/devspace
Length of output: 341
🌐 Web query:
Claude Prompt Caching total input tokens uncached input cache read cache creation documentation
💡 Result:
Claude prompt caching allows you to store and reuse portions of your prompt (the "prefix") across multiple API requests, significantly reducing latency and cost [1][2]. The total input tokens for a request using prompt caching are calculated as the sum of three distinct categories provided in the usage statistics [3][4]: total_input_tokens = cache_read_input_tokens + cache_creation_input_tokens + input_tokens Here is the breakdown of these terms: 1. cache_read_input_tokens: The number of tokens retrieved from an existing cache entry [3][5]. These are billed at 10% of the standard input token price [3][1]. 2. cache_creation_input_tokens: The number of tokens being written to a new cache entry (or a refresh) [3][5]. These are billed at a premium (1.25x the base input price for a 5-minute TTL, or 2x for a 1-hour TTL) [1][6]. 3. input_tokens: The number of "uncached" tokens, representing only the content that appears after the last cache breakpoint in your request [3][4]. These are billed at the standard input token rate [3][1]. Key operational details: - Prefix Matching: Caching is based on exact "prefix matching" [7][8]. The API caches content from the start of the request up to designated cache breakpoints [5][7]. If any part of the prompt before a breakpoint changes, the cache is invalidated for that breakpoint and all subsequent ones [9][4][8]. - Diagnostics: If you observe cache_read_input_tokens dropping to zero unexpectedly, you can use cache diagnostics to determine if the prefix has changed or if the cache entry has expired [9]. - Ordering Matters: Because the cache is a byte-for-byte prefix match, the order of content—such as tool definitions, system prompts, and message history—is critical [7][8]. Organizing your prompt to maximize stable, shared prefixes is the most effective way to utilize this feature [7].
Citations:
- 1: https://platform.claude.com/docs/en/about-claude/pricing?f80ce999_sort_date=desc
- 2: https://claude.com/blog/prompt-caching
- 3: https://platform.claude.com/docs/en/build-with-claude/prompt-caching
- 4: https://github.com/anthropics/skills/blob/main/skills/claude-api/shared/prompt-caching.md
- 5: https://code.claude.com/docs/en/prompt-caching.md
- 6: https://platform.claude.com/docs/en/about-claude/pricing?fcdaa149_sort_date=desc&refid=e08b4151-8993-4b50-839f-e4d6991e2e07
- 7: https://claude.com/blog/lessons-from-building-claude-code-prompt-caching-is-everything
- 8: https://callsphere.ai/blog/how-claude-prompt-caching-works-internals-and-architecture
- 9: https://platform.claude.com/docs/en/build-with-claude/cache-diagnostics
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="$(fd 'local-agent-adapters\.ts$' . | head -n 1)"
printf 'File: %s\n' "$file"
printf '\nLines 180-205:\n'
awk 'NR>=180 && NR<=205 {printf "%6d %s\n", NR, $0}' "$file"
printf '\nLines 220-290:\n'
awk 'NR>=220 && NR<=290 {printf "%6d %s\n", NR, $0}' "$file"
printf '\nSearch relevant usages:\n'
rg -n "totalTokens|cache_read_input_tokens|cache_creation_input_tokens|inputTokens|cachedInputTokens|cacheCreationInputTokens|nonNegativeInteger|persist|observer" src || trueRepository: Waishnav/devspace
Length of output: 16327
Include Claude cache tokens in totalTokens.
totalTokens currently sums only inputTokens and outputTokens, then persists the value through the workflow agent observer. Claude prompt-cached requests define total input tokens as input_tokens + cache_read_input_tokens + cache_creation_input_tokens, so adjust the aggregate to include the cached token fields before storing it.
🧰 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 188 - 198, Update the token
aggregate in the usage return block to include cacheReadInputTokens and
cacheCreationInputTokens alongside inputTokens and outputTokens when calculating
totalTokens. Preserve the existing non-negative normalization and undefined
handling in the surrounding token fields.
1 participant
This layer adds a provider-neutral observation contract and wires native Codex and Claude SDK events into workflow persistence. Session ids and normalized activity are written immediately; partial token snapshots are throttled to at most once every five seconds and final usage is written immediately. Usage from schema-enforcement retries is accumulated under the same workflow call, and the observer always flushes before a provider run exits.\n\nDepends on #152.
Summary by CodeRabbit