feat(workflow): add durable observability contracts - #156
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 agent observations and token usage types, migration 9 storage, and WorkflowStore APIs for persisting, aggregating, validating, and retrieving ordered observations. ChangesWorkflow observability
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant AgentCaller
participant WorkflowStore
participant workflow_agent_calls
participant workflow_agent_observations
AgentCaller->>WorkflowStore: completeAgentCall(usage)
WorkflowStore->>workflow_agent_calls: write usageJson and finalUsageJson
AgentCaller->>WorkflowStore: appendAgentObservation(observation)
WorkflowStore->>workflow_agent_observations: insert sequenced observation
WorkflowStore->>workflow_agent_calls: update aggregated usage
AgentCaller->>WorkflowStore: listAgentObservations(runId, callIndex)
WorkflowStore->>workflow_agent_observations: query ordered observations
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 SummaryAdds provider-neutral workflow observability storage contracts and token-usage tracking.
Confidence Score: 4/5The PR appears safe to merge, with a non-blocking issue in how oversized observation JSON is truncated. The observability schema and usage persistence are coherent, but oversized dataJson payloads are stored as invalid JSON because the new path performs raw byte truncation instead of JSON-aware truncation. Files Needing Attention: src/workflow-store.ts
|
| Filename | Overview |
|---|---|
| src/db/migrations.ts | Adds migration 9 with usage columns and the observation table, indexes, and run-level cascade. |
| src/db/schema.ts | Models agent usage fields and the workflow observation storage schema. |
| src/local-agent-observations.ts | Defines provider-neutral activity and token-usage contracts plus normalization and merge helpers. |
| src/workflow-store.ts | Adds observation persistence and usage retrieval, but raw truncation can invalidate oversized dataJson payloads. |
| src/workflow-types.ts | Exposes observation records, token-usage fields, and the observation payload cap. |
| src/workflow-store.test.ts | Covers basic activity persistence and usage propagation to agent-call records. |
Flowchart
%%{init: {'theme': 'neutral'}}%%
flowchart LR
P[Local agent provider] --> O[LocalAgentObservation]
O --> A[WorkflowStore.appendAgentObservation]
A --> T[(workflow_agent_observations)]
A -->|usage observation| C[(workflow_agent_calls.usage_json)]
F[completeAgentCall] -->|final usage| C
Reviews (1): Last reviewed commit: "feat(workflow): add observability storag..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| : null; | ||
| const dataJson = input.dataJson | ||
| ? truncateText(input.dataJson, WORKFLOW_LIMITS.observationDataJsonBytes, "dataJson") |
There was a problem hiding this comment.
Oversized dataJson values are truncated at an arbitrary byte boundary, leaving stored diagnostics with invalid JSON or a replacement character when a multibyte character is split. Use JSON-aware truncation so consumers can reliably parse the capped payload.
|
@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: 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/workflow-store.ts`:
- Around line 898-945: Update appendAgentObservation to read the existing
usage_json for the matching workflow_agent_calls row within the transaction,
merge it with observation.usage when processing usage observations, and persist
the merged usage while leaving the inserted observation row unchanged. Add a
regression test covering two partial usage observations with different token
fields and verify the agent call contains both fields.
- Around line 1308-1313: Update truncateText to avoid decoding a partial UTF-8
code point: after slicing the first maxBytes bytes, back up over trailing UTF-8
continuation bytes before converting the buffer to a string. Ensure the returned
truncated value never exceeds maxBytes while preserving the existing error
behavior for an empty result.
🪄 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: 671f75bb-8e05-4b70-8a76-f2edeed7c5db
📒 Files selected for processing (7)
src/db/migrations.tssrc/db/schema.tssrc/local-agent-observations.tssrc/oauth-store.test.tssrc/workflow-store.test.tssrc/workflow-store.tssrc/workflow-types.ts
| appendAgentObservation(input: AppendAgentObservationInput): WorkflowAgentObservationRecord { | ||
| const call = this.requireAgentCall(input.runId, input.callIndex); | ||
| const observation = input.observation; | ||
| const usageJson = observation.kind === "usage" | ||
| ? JSON.stringify(observation.usage) | ||
| : null; | ||
| const dataJson = input.dataJson | ||
| ? truncateText(input.dataJson, WORKFLOW_LIMITS.observationDataJsonBytes, "dataJson") | ||
| : null; | ||
| const createdAt = isoNow(); | ||
| const transaction = this.database.sqlite.transaction(() => { | ||
| const next = this.database.sqlite | ||
| .prepare( | ||
| `select coalesce(max(seq), 0) + 1 as next_seq | ||
| from workflow_agent_observations where run_id = ? and call_index = ?`, | ||
| ) | ||
| .get(input.runId, input.callIndex) as { next_seq: number }; | ||
| this.database.sqlite | ||
| .prepare( | ||
| `insert into workflow_agent_observations ( | ||
| run_id, call_index, seq, provider, kind, activity_id, message, | ||
| tool_name, tool_status, detail, usage_json, data_json, created_at | ||
| ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, | ||
| ) | ||
| .run( | ||
| input.runId, | ||
| input.callIndex, | ||
| next.next_seq, | ||
| call.provider, | ||
| observation.kind, | ||
| observation.kind === "activity" ? observation.activityId ?? null : null, | ||
| observation.kind === "activity" ? observation.message ?? null : null, | ||
| observation.kind === "activity" ? observation.toolName ?? null : null, | ||
| observation.kind === "activity" ? observation.toolStatus ?? null : null, | ||
| observation.kind === "activity" ? observation.detail ?? null : null, | ||
| usageJson, | ||
| dataJson, | ||
| createdAt, | ||
| ); | ||
| if (usageJson) { | ||
| this.database.sqlite | ||
| .prepare( | ||
| `update workflow_agent_calls | ||
| set usage_json = ?, updated_at = ? | ||
| where run_id = ? and call_index = ?`, | ||
| ) | ||
| .run(usageJson, createdAt, input.runId, input.callIndex); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Merge partial usage observations before updating the agent call.
Lines 937-944 replace usage_json with the newest payload. This drops fields from earlier partial observations. Read the current usage inside the transaction, merge it with observation.usage, and persist the merged value. Keep the individual observation row unchanged.
Add a regression test with two partial usage observations that contain different token fields.
Based on PR objectives, the store must aggregate usage observations.
🤖 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-store.ts` around lines 898 - 945, Update appendAgentObservation
to read the existing usage_json for the matching workflow_agent_calls row within
the transaction, merge it with observation.usage when processing usage
observations, and persist the merged usage while leaving the inserted
observation row unchanged. Add a regression test covering two partial usage
observations with different token fields and verify the agent call contains both
fields.
| function truncateText(value: string, maxBytes: number, label: string): string { | ||
| if (Buffer.byteLength(value, "utf8") <= maxBytes) return value; | ||
| const truncated = Buffer.from(value, "utf8").subarray(0, maxBytes).toString("utf8"); | ||
| if (!truncated) throw new Error(`${label} exceeds limit (${maxBytes} bytes)`); | ||
| return truncated; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Preserve the byte limit at UTF-8 boundaries.
Line 1310 can decode a partial code point as U+FFFD. The replacement character can make the returned value exceed maxBytes. Back up over UTF-8 continuation bytes before decoding the slice.
Proposed fix
function truncateText(value: string, maxBytes: number, label: string): string {
if (Buffer.byteLength(value, "utf8") <= maxBytes) return value;
- const truncated = Buffer.from(value, "utf8").subarray(0, maxBytes).toString("utf8");
+ const bytes = Buffer.from(value, "utf8");
+ let end = Math.min(maxBytes, bytes.length);
+ while (end > 0 && ((bytes[end] ?? 0) & 0b1100_0000) === 0b1000_0000) end -= 1;
+ const truncated = bytes.subarray(0, end).toString("utf8");
if (!truncated) throw new Error(`${label} exceeds limit (${maxBytes} bytes)`);
return truncated;
}Based on PR objectives, observation diagnostic data is size-capped.
🤖 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-store.ts` around lines 1308 - 1313, Update truncateText to avoid
decoding a partial UTF-8 code point: after slicing the first maxBytes bytes,
back up over trailing UTF-8 continuation bytes before converting the buffer to a
string. Ensure the returned truncated value never exceeds maxBytes while
preserving the existing error behavior for an empty result.
|
Closing this stack in favor of the alternative observability implementation. Review surfaced unresolved storage-contract issues here, including invalid JSON truncation and retention semantics that did not actually bound persisted observations. The provider-neutral shape is still a useful design reference, but this PR will not be carried forward. |
1 participant
Workflow runs need a durable, bounded record of what each agent call did and how much model usage it consumed so the CLI and TUI can inspect completed and in-progress work. This bottom layer adds the shared observation and token-usage contracts, an additive SQLite migration, and store operations for ordered call observations plus latest/final usage. Observation payloads are optional and size-capped; providers that do not report a value remain unset.
Summary by CodeRabbit
New Features
Tests