feat(workflow): persist provider observations - #157
Conversation
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughThe change adds provider-neutral observation callbacks and token-usage reporting. Workflow execution aggregates and throttles usage observations, persists them in the journal, and records final usage for successful or failed agent calls. ChangesObservation and usage reporting
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Provider
participant WorkflowWorker
participant WorkflowAPI
participant WorkflowJournal
Provider->>WorkflowWorker: Emit observations and return usage
WorkflowWorker->>WorkflowAPI: Forward observations and usage
WorkflowAPI->>WorkflowJournal: Append observations
WorkflowAPI->>WorkflowJournal: Complete agent call with aggregated 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 |
Greptile SummaryThis PR carries provider activity and token-usage observations through the local runtime and workflow worker, persists observations in the workflow journal, and attaches final usage to completed agent calls.
Confidence Score: 4/5The PR should not merge until observation persistence failures are prevented from converting provider results into failures or leaving agent calls unfinished. The new synchronous observation callback performs fallible journal writes without the best-effort error boundary used by the runtime, and the same unguarded flush precedes failure finalization. Files Needing Attention: src/workflow-api.ts
|
| Filename | Overview |
|---|---|
| src/workflow-api.ts | Adds observation persistence and final-usage handling, but unguarded observation writes can fail provider calls and prevent journal finalization. |
| src/local-agent-runtime.ts | Extends runtime inputs and results with observations and usage while providing a best-effort callback helper. |
| src/workflow-worker.ts | Correctly forwards observation callbacks and provider usage between the workflow API and local runtime. |
| src/workflow-engine.test.ts | Covers successful observation and usage persistence but does not exercise journal-write failures. |
Sequence Diagram
sequenceDiagram
participant E as Workflow API
participant J as Workflow Journal
participant P as Provider
E->>J: startAgentCall
E->>P: run(onObservation)
P-->>E: activity / usage observation
E->>J: appendAgentObservation
P-->>E: provider result + usage
E->>J: flush pending usage
E->>J: completeAgentCall(finalUsage)
Note over E,J: An append exception currently escapes and can interrupt completion/failure finalization
Reviews (1): Last reviewed commit: "feat(workflow): persist provider observa..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| deps.journal.appendAgentObservation({ | ||
| runId: deps.runId, | ||
| callIndex: index, | ||
| observation: { kind: "usage", usage }, | ||
| }); |
There was a problem hiding this comment.
When the journal throws while appending an activity or usage observation, the exception escapes the provider callback and discards an otherwise successful provider result. The same unguarded write in flushUsage runs before failAgentCall, so a flush failure leaves the agent-call record unfinished.
|
@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: 3
🤖 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-runtime.ts`:
- Around line 39-40: Update the observation dispatch using onObservation so
returned promises are handled and rejected callbacks are suppressed, while
preserving handling of synchronous throws and the existing optional-callback
behavior.
- Around line 50-51: Update the live runtime adapters for Codex, Claude,
Opencode, Pi, and ACP to invoke notifyLocalAgentObservation for activity and
usage events during execution, rather than only forwarding onObservation to the
API. Ensure emitted usage uses LocalAgentTokenUsage and is persisted by the
worker, while preserving the existing finalResponse, items, and usage() results.
In `@src/workflow-api.ts`:
- Around line 458-471: Update the schema-enforcement flow around
enforceAgentSchema and the retry-exhaustion path before failAgentCall to merge
usage from each completed latestProviderResult into the usage journal, ensuring
the latest provider usage is flushed even when enforcement throws. Add a
retry-exhaustion test where the provider returns usage without emitting a usage
observation, and verify that usage is persisted.
🪄 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: 895c0708-343a-4b17-abff-bb1537aa82f1
📒 Files selected for processing (4)
src/local-agent-runtime.tssrc/workflow-api.tssrc/workflow-engine.test.tssrc/workflow-worker.ts
| /** Receives provider-neutral activity and usage observations while running. */ | ||
| onObservation?: (observation: LocalAgentObservation) => void; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline src/local-agent-runtime.ts --items all --type interface,function,class
rg -n -C 5 'onObservation|notifyLocalAgentObservation' \
src/local-agent-runtime.ts src/local-agent-runtime.test.tsRepository: Waishnav/devspace
Length of output: 2904
🏁 Script executed:
#!/bin/bash
set -euo pipefail
node - <<'JS'
let unhandled
process.on('unhandledRejection', (reason) => {
unhandled = reason
process.exitCode = 1
})
function notifyLocalAgentObservation(onObservation, observation) {
try {
onObservation?.(observation)
} catch {
// Observability is best effort and must not turn a provider result into a failure.
}
}
async function rejected() {
return Promise.reject(new Error('observer rejected'))
}
notifyLocalAgentObservation(rejected, { event: 'x' })
setTimeout(() => {
console.log(unhandled ? `unhandled rejection: ${unhandled.message}` : 'no unhandled rejection')
}, 10)
JSRepository: Waishnav/devspace
Length of output: 194
Suppress rejected asynchronous observer callbacks.
input.onObservation?.(observation) can return a Promise. The current try/catch only handles synchronous throws, so an async observer rejection escapes unhandled. Catch observer promises after invoking the callback.
🤖 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 39 - 40, Update the observation
dispatch using onObservation so returned promises are handled and rejected
callbacks are suppressed, while preserving handling of synchronous throws and
the existing optional-callback behavior.
| /** Latest provider-reported usage, when available. */ | ||
| usage?: LocalAgentTokenUsage; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline src/local-agent-runtime.ts --items all --type class,function
rg -n -C 6 'notifyLocalAgentObservation|onObservation|usage:' src --glob '*local-agent*.ts'Repository: Waishnav/devspace
Length of output: 5270
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== local-agent files =="
git ls-files 'src/**/local-agent*.ts'
echo
echo "== outline src/local-agent-runtime.ts =="
ast-grep outline src/local-agent-runtime.ts --items all --type class,function || true
echo
echo "== relevant src/local-agent-runtime.ts sections =="
sed -n '1,190p' src/local-agent-runtime.ts | cat -n
echo
echo "== local-agent observations =="
sed -n '1,220p' src/local-agent-observations.ts | cat -n
echo
echo "== imports/usages of notifyLocalAgentObservation and onObservation =="
rg -n -C 4 'notifyLocalAgentObservation|onObservation|LocalAgentObservation|usageAsObservation|normalizeLocalAgentTokenUsage|LocalAgentRunResult' src --glob '*.ts'Repository: Waishnav/devspace
Length of output: 25055
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== local-agent adapters relevant sections =="
sed -n '60,270p' src/local-agent-adapters.ts | cat -n
sed -n '400,490p' src/local-agent-adapters.ts | cat -n
echo
echo "== workflow API persistObservation section =="
sed -n '320,490p' src/workflow-api.ts | cat -n
echo
echo "== references to notifyLocalAgentObservation in src =="
rg -n 'notifyLocalAgentObservation' src --glob '*.ts'
echo
echo "== imports in local-agent runtime and adapters =="
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()
print(f"== imports in {path} ==")
import re
for m in re.finditer(r"^(?:import\s+(?:type\s+)?(?:.*?\n)*?from ['\"]([^'\"]+)['\"])|(^export\s+\{(?:.*?\n)*?from ['\"]([^'\"]+)['\"])", text, re.M):
print(text[m.start():m.start()+400].split('\n')[0])
PYRepository: Waishnav/devspace
Length of output: 19853
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== all adapters and their result statements =="
python3 - <<'PY'
from pathlib import Path
p=Path('src/local-agent-adapters.ts')
text=p.read_text()
lines=text.splitlines()
for i,l in enumerate(lines,1):
if 'class ' in l and 'Adapter' in l:
print(f'== {p}:{i} ==')
c1='async run'
c2=' async run'
if (c1 in l or c2 in l):
start=max(1,i-3)
# find start of run by walking backwards to class
for j in range(i-1,0,-1):
if lines[j-1].startswith('class '):
start=j
break
end=None
brace=0
inBody=False
for k in range(i-1, len(lines)):
if ('async run' in lines[k] or (k>i and lines[k].strip().startswith('async run'))) and not inBody:
inBody=True
if inBody:
brace += lines[k].count('{') - lines[k].count('}')
if brace == 0 and ('return {' in lines[k] or 'return {' in l for l in [lines[k]]):
end=k
break
if end is None:
for j in range(k+1, min(len(lines), i+80)):
if 'onObservation' in lines[j] or 'notifyLocalAgent' in lines[j]:
end=j
break
if end is None:
end=min(len(lines), k+120)
print('\n'.join(f'{n}: {lines[n]}' for n in range(start, end+1)))
print()
break
PY
echo
echo "== inspect all adapter run return blocks =="
python3 - <<'PY'
from pathlib import Path
import re
text=Path('src/local-agent-adapters.ts').read_text()
for name, pattern in [
('Codex', r'(?m)^class CodexLocalAgentAdapter[\s\S]*?return \{[\s\S]*?\}\s*\n'),
('Claude', r'(?m)^class ClaudeLocalAgentAdapter[\s\S]*?return \{[\s\S]*?\}\s*\n'),
('OpenCode', r'(?m)^class OpencodeLocalAgentAdapter[\s\S]*?return \{[\s\S]*?\}\s*\n'),
('AcpLocal', r'(?m)^class AcpLocalAgentAdapter[\s\S]*?return \{[\s\S]*?\}\s*\n'),
('PiRpc', r'(?m)^class PiRpcLocalAgentAdapter[\s\S]*?return \{[\s\S]*?\}\s*\n'),
]:
m=re.search(pattern, text)
print(f'== {name} ==')
print(m.group(0) if m else "no match")
PY
echo
echo "== workflow API non-schema final path =="
sed -n '160,185p' src/workflow-api.ts | cat -nRepository: Waishnav/devspace
Length of output: 12002
Emit local-agent observations from the live adapters.
The packaged providers currently only pass onObservation through to the API, but notifyLocalAgentObservation is never called and the runtime adapters return only finalResponse, items, and usage() at the end. Add activity/usage emissions for the supported path, including Codex, Claude, Opencode, Pi, and the ACP adapters, so the worker can persist them.
🤖 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 50 - 51, Update the live runtime
adapters for Codex, Claude, Opencode, Pi, and ACP to invoke
notifyLocalAgentObservation for activity and usage events during execution,
rather than only forwarding onObservation to the API. Ensure emitted usage uses
LocalAgentTokenUsage and is persisted by the worker, while preserving the
existing finalResponse, items, and usage() results.
Source: Coding guidelines
| let latestProviderResult: WorkflowProviderRunResult | undefined; | ||
| const enforced = await enforceAgentSchema({ | ||
| schema: agentOpts.schema, | ||
| prompt: providerPrompt, | ||
| provider, | ||
| run: (p, options) => | ||
| deps.runProvider({ | ||
| run: async (p, options) => { | ||
| latestProviderResult = await deps.runProvider({ | ||
| ...providerBase, | ||
| prompt: p, | ||
| providerSessionId: options.providerSessionId, | ||
| ...(options.mode === "native" ? { schema: agentOpts.schema } : {}), | ||
| }), | ||
| }); | ||
| return latestProviderResult; | ||
| }, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Persist provider-result usage when schema enforcement fails.
Lines 464-470 retain latestProviderResult but do not merge its usage before schema enforcement can throw. On retry exhaustion, execution goes to Line 563, where flushUsage() persists only callback-derived usage. failAgentCall has no usage field.
If a provider reports usage only in its result, failed schema calls lose that usage. Merge result usage for each completed attempt and journal the latest usage before failAgentCall. Add a retry-exhaustion test that returns usage without emitting a usage observation.
Also applies to: 563-563
🤖 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/workflow-api.ts` around lines 458 - 471, Update the schema-enforcement
flow around enforceAgentSchema and the retry-exhaustion path before
failAgentCall to merge usage from each completed latestProviderResult into the
usage journal, ensuring the latest provider usage is flushed even when
enforcement throws. Add a retry-exhaustion test where the provider returns usage
without emitting a usage observation, and verify that usage is persisted.
|
Closing this stack in favor of the alternative observability implementation. Review found lifecycle gaps in the persistence path, especially that observation/journal write failures can escape the best-effort telemetry boundary and interfere with provider-call finalization. The overall callback design had good ideas, but this version is not the one we are carrying forward. |
1 participant
Provider runtimes need a way to report activity and usage without coupling adapters to SQLite. The workflow journal now accepts provider observations, appends activity immediately, coalesces usage snapshots on a five-second cadence, flushes pending usage when a call finishes or fails, and records the provider result as final usage. This layer depends on the observability contracts below it.
Summary by CodeRabbit
New Features
Bug Fixes