mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-30 23:14:37 +00:00
[codex] feat(c4): add journal progress projection shadow path (#4535)
This commit is contained in:
Steven Enamakel
@@ -0,0 +1,193 @@
|
||||
# C4 — Journal-backed progress projection & `progress_tracing` deletion
|
||||
|
||||
Status: execution plan (2026-07-04), written after a ground-truth code map of
|
||||
both the OpenHuman progress surface and the vendored crate observability
|
||||
primitives. This is the actionable plan for the C4 workstream's July 2026
|
||||
continuation notes, and it is the gated prerequisite for **doc 03 (V3) Step
|
||||
5** — deleting the `ProviderDelta` bridge and `progress_tracing`.
|
||||
|
||||
## 1. Corrected architecture (what the map found)
|
||||
|
||||
- **There is no crate `SpanCollector`.** The only span state machine is
|
||||
OpenHuman's `SpanCollector` in `src/openhuman/agent/progress_tracing.rs`
|
||||
(1272 lines). The crate does **not** build spans — it journals raw
|
||||
`AgentObservation`s and lets exporters project.
|
||||
- **The journal/status/persistence stack already exists and is attached to
|
||||
every run.** `run_turn_via_tinyagents_shared`
|
||||
(`src/openhuman/tinyagents/mod.rs:420`) mints a run id, seeds the `EventSink`
|
||||
with it, and `attach_turn_journal` (`src/openhuman/tinyagents/journal.rs:304`)
|
||||
installs `StoreEventJournal` (over `JsonlAppendStore`) via a
|
||||
`JournalSink → RedactingSink → FanOutSink`, plus a durable `FileStatusStore`.
|
||||
Every run already durably records the crate `AgentEvent` stream as
|
||||
`AgentObservation`s.
|
||||
- **Two producers of `AgentProgress`:**
|
||||
- Crate path: `OpenhumanEventBridge` (`src/openhuman/tinyagents/observability.rs:464`)
|
||||
maps `AgentEvent` → `AgentProgress` live (stateful: iteration cursor,
|
||||
subagent `scope`, `tool_names` recovery, display labels, failure class).
|
||||
- Legacy path: `session/tool_progress.rs` `TurnProgress` +
|
||||
`spawn_delta_forwarder` maps engine callbacks + `ProviderDelta` →
|
||||
`AgentProgress` (the **Step-5 deletion target**, 253 lines).
|
||||
- **`SpanCollector` consumes `AgentProgress`** (the bridge *output*), while the
|
||||
journal captures the `AgentEvent` *input*. The web progress bridge
|
||||
(`channels/providers/web/progress_bridge.rs:157`) is a side-observer of the
|
||||
`AgentProgress` channel: `collector.record(&event, now)` per event, then
|
||||
`collector.finish()` + `export_run_trace(config, spans)` at loop exit.
|
||||
- **Langfuse is exported twice, in two data models.**
|
||||
`progress_tracing/langfuse.rs` (825 lines) hand-rolls a `TraceSpan`→Langfuse
|
||||
ingestion batch; the crate `observability/langfuse/` already builds the same
|
||||
batch from `&[AgentObservation]` (`LangfuseClient::send_observations`,
|
||||
`build_ingestion_batch`). Same proxy path, same `207` handling.
|
||||
|
||||
## 2a. BLOCKER found during S1 (2026-07-04): the mapping is not journal-replayable
|
||||
|
||||
The S1 attempt to extract a pure `event_to_progress` revealed that the live
|
||||
`OpenhumanEventBridge` mapping **depends on live side-channels absent from the
|
||||
journal**, so "parity by construction via journal replay" (§2 below) does
|
||||
**not** hold as written:
|
||||
|
||||
- **`ToolCompleted`** (`observability.rs:745`): the crate event
|
||||
`ToolCompleted { call_id, tool_name, started_at_ms, input, output }` carries
|
||||
**no outcome**. `success` / `elapsed_ms` / `output_chars` / failure class are
|
||||
read from `self.failure_map`, a `call_id → (ok, failure, elapsed, chars)` map
|
||||
populated by `ToolOutcomeCaptureMiddleware` — not journalled.
|
||||
- **`record_usage`** (`observability.rs:298`): drains `usage_carry` (the model
|
||||
adapter's provider `UsageInfo` FIFO) to restore **charged USD**, cache-creation
|
||||
and context-window tokens the crate `Usage` drops. Not journalled.
|
||||
|
||||
A journal-only projection therefore yields structurally-correct spans with
|
||||
**degraded attributes** (assumed-success tools, zero durations/sizes, `$0`
|
||||
cost) — which cannot pass the S3/S5 parity gate.
|
||||
|
||||
### Corrected prerequisite — S0: make the journal self-sufficient (crate work)
|
||||
|
||||
Before any journal projection can hit parity, the enriching data must live in
|
||||
the journalled `AgentEvent`s, not in OpenHuman side-channels:
|
||||
|
||||
1. **Enrich `AgentEvent::ToolCompleted`** in the crate with the outcome:
|
||||
`success: bool`, `duration_ms`, `output_bytes` (and an optional structured
|
||||
failure), populated in `tools.rs::finish_tool_call` from the `ToolResult`
|
||||
and the `started_at_ms` it already tracks. OpenHuman's bridge then reads them
|
||||
from the event; `failure_map`/`ToolOutcomeCaptureMiddleware` shrink to the
|
||||
product-specific `ToolFailureClass` mapping (or that too moves onto the
|
||||
event). This is a V-series crate PR (additive event fields).
|
||||
2. **Usage accounting**: decide whether charged-USD/provider-cost belongs on a
|
||||
crate event (e.g. a `provider_cost`/`cache_creation` extension on
|
||||
`UsageRecorded`/`Usage`) or stays a documented, accepted OpenHuman-only trace
|
||||
attribute filled at export time from the status/cost store rather than the
|
||||
live side-channel. (The cost roll-up is already persisted per-run; the trace
|
||||
can read it from there instead of `usage_carry`.)
|
||||
3. Only after S0 does §2's "reuse the mapping" become true by construction.
|
||||
|
||||
**Sequencing impact:** S0 (crate event enrichment) is now the first slice and
|
||||
gates S1→S2. It is separate crate work that should be PR'd upstream like V1/V2.
|
||||
The remaining S1–S6 below stand, rebased on S0.
|
||||
|
||||
## 2. The parity-preserving approach (valid only after S0)
|
||||
|
||||
Reproduce the span tree **by construction**, not by re-deriving it:
|
||||
|
||||
> Replay journal `AgentObservation`s → `AgentProgress` using the *same* mapping
|
||||
> the live bridge uses → fold through the *existing* `SpanCollector`.
|
||||
|
||||
Concretely:
|
||||
|
||||
1. **Extract the bridge mapping into a pure, reusable function**
|
||||
`event_to_progress(event: &AgentEvent, state: &mut BridgeState, scope) -> Vec<AgentProgress>`
|
||||
(state = iteration cursor + `tool_names` + scope). The live
|
||||
`OpenhumanEventBridge::on_event` becomes a thin driver over it (refactor, **no
|
||||
behavior change** — guarded by the existing bridge tests). This makes the
|
||||
AgentEvent→AgentProgress mapping a single source of truth.
|
||||
2. **Journal projection**
|
||||
`spans_from_observations(ctx: TraceContext, obs: &[AgentObservation]) -> Vec<TraceSpan>`:
|
||||
fold each observation through `event_to_progress` into `AgentProgress`, feed a
|
||||
fresh `SpanCollector::record(...)` stamped with `obs.ts_ms`, then `finish()`.
|
||||
Because it reuses `SpanCollector`, span-shape parity is guaranteed for every
|
||||
`AgentProgress` variant the journal can produce.
|
||||
3. **Parity harness** (`progress_tracing`'s `tests.rs` is the oracle): drive a
|
||||
representative synthetic run and assert
|
||||
`spans_from_observations(journal)` == `SpanCollector` fed the live
|
||||
`AgentProgress`. Cover: multi-iteration turn, tool calls (success + failed +
|
||||
unknown-tool recovery), model-call generation spans w/ usage & reasoning &
|
||||
cache-creation, nested sub-agents, cost roll-up.
|
||||
|
||||
### Parity gaps to resolve (AgentProgress with no journal `AgentEvent` source)
|
||||
|
||||
`SpanCollector` ignores streaming/content deltas for spans
|
||||
(`progress_tracing.rs:1076`), so `TextDelta`/`ThinkingDelta`/`ToolCallArgsDelta`
|
||||
**do not affect span parity** — safe to drop on replay. The variants that *do*
|
||||
carry span data but have no direct `AgentEvent`:
|
||||
|
||||
| AgentProgress | Span effect | Journal source | Resolution |
|
||||
| --- | --- | --- | --- |
|
||||
| `TurnContent{prompt, reply}` | root span `input`/`output` (gated on `capture_content`) | `ModelCompleted.input/output` present in journal but shaped differently | project root i/o from `ModelStarted`/`ModelCompleted` payloads, or emit a journalled content event |
|
||||
| `TaskBoardUpdated` | none (no span) | n/a | ignore for spans |
|
||||
| `SubagentAwaitingUser` | subagent span attr | partial (`SubAgentStarted/Completed` only) | accept minor attr gap or add a crate event (V6) |
|
||||
| `TurnCostUpdated` | root usage roll-up | `UsageRecorded`/`CostRecorded` in journal | project from those (already in bridge) |
|
||||
|
||||
Document any accepted gap in the parity test as an explicit, reviewed
|
||||
exception.
|
||||
|
||||
## 3. Langfuse swap (separable, lower-risk slice)
|
||||
|
||||
Replace `progress_tracing/langfuse.rs::push_spans` with the crate
|
||||
`LangfuseClient::proxy(...).send_observations(trace_cfg, &obs)` reading the run's
|
||||
journal. Parity test: assert the crate batch matches the existing batch shape
|
||||
for a fixture run (trace-create + per-observation generation/span/event,
|
||||
`usageDetails`/`costDetails`, `207` handling, content gating from
|
||||
`agent_tracing.capture_content`). This deletes ~825 lines independently of the
|
||||
span-projection slice.
|
||||
|
||||
## 4. Sequenced slices (each: code + tests + green build; deletions gated)
|
||||
|
||||
0. **S0 — enrich crate `AgentEvent::ToolCompleted`** with `duration_ms` /
|
||||
`output_bytes` / `error` so the journal is self-sufficient for tool outcomes
|
||||
(§2a). **DONE** — tinyagents#18 (branch `feat/tool-completed-outcome`,
|
||||
933 tests green). Also enriches the crate Langfuse tool span. Merge + gitlink
|
||||
bump precede S1. (Usage/charged-USD accounting — §2a item 2 — is still open:
|
||||
fill it at export time from the persisted per-run cost store rather than the
|
||||
`usage_carry` side-channel.)
|
||||
1. **S1 — extract `event_to_progress`** (pure mapping) + make the live bridge a
|
||||
driver over it. No behavior change; existing bridge/`observability` tests are
|
||||
the gate. *No deletion.*
|
||||
2. **S2 — `spans_from_observations`** journal projection + parity harness vs
|
||||
`SpanCollector`. **IN PROGRESS** — additive projection module now covers the
|
||||
single-agent spine, S0 tool outcomes, root `TurnContent` from captured model
|
||||
I/O, and sub-agent lifecycle/scoped child tool/model spans. Remaining known
|
||||
gap: per-call charged USD/cache-creation is still zero until export-time cost
|
||||
store reconciliation lands.
|
||||
3. **S3 — flip the web progress bridge** to build spans from the journal at run
|
||||
end (`spans_from_observations`) instead of the live `AgentProgress`
|
||||
side-observer; keep the old path behind a shadow-compare for one release
|
||||
(log divergences), matching the C1 `session_shadow_reads` pattern. **STARTED**
|
||||
— the web bridge now keeps live export as-is and logs a structural
|
||||
journal-projection shadow comparison keyed by the durable journal run id.
|
||||
4. **S4 — Langfuse swap** to the crate exporter (§3); delete
|
||||
`progress_tracing/langfuse.rs` (~825 + tests).
|
||||
5. **S5 — delete `progress_tracing.rs` + `SpanCollector`** once S3 shadow shows
|
||||
no divergence for one release **and** V3 projection parity holds (the doc 07
|
||||
gate). Tick the deletion ledger (~2k incl. tests).
|
||||
6. **S6 (= V3 Step 5)** — with `AgentProgress` now a journal projection, delete
|
||||
the legacy `ProviderDelta` bridge in `session/tool_progress.rs` (253) and the
|
||||
`spawn_delta_forwarder`, since the crate event path is the only producer.
|
||||
|
||||
## 5. Gates (doc 07)
|
||||
|
||||
- `progress_tracing` delete: **C4 shadow parity (S3) for one release AND V3
|
||||
projection parity.** Langfuse (S4) may land earlier (independent shape parity).
|
||||
- Approval/security/redaction boundaries unchanged: the journal path already
|
||||
runs through `RedactingSink` (`journal.rs:327`); the projection must not
|
||||
re-introduce raw prompt/PII into spans beyond what `capture_content` already
|
||||
gates.
|
||||
|
||||
## 6. Key file references
|
||||
|
||||
Producer/bridge: `tinyagents/observability.rs:464` (`OpenhumanEventBridge`),
|
||||
`tinyagents/journal.rs:304` (`attach_turn_journal`), `tinyagents/mod.rs:420`
|
||||
(`run_turn_via_tinyagents_shared`). Span machine:
|
||||
`agent/progress_tracing.rs` (`SpanCollector:307`, `record:686`, `finish:1087`,
|
||||
`export_run_trace:1254`), driven at
|
||||
`channels/providers/web/progress_bridge.rs:157`. Crate foundation:
|
||||
`harness/observability/{mod,types}.rs` (`HarnessEventJournal:156`,
|
||||
`StoreEventJournal`, `JournalSink:581`, `AgentObservation:40`),
|
||||
`harness/observability/langfuse/`. Parity oracle:
|
||||
`agent/progress_tracing/tests.rs` (1169 lines).
|
||||
@@ -60,9 +60,14 @@ use crate::openhuman::agent::progress::AgentProgress;
|
||||
use crate::openhuman::config::schema::{AgentTracingBackend, AgentTracingConfig};
|
||||
use crate::openhuman::config::Config;
|
||||
|
||||
/// Journal-backed projection from durable tinyagents observations.
|
||||
pub(crate) mod journal_projection;
|
||||
/// Langfuse ingestion exporter (remote push to the co-hosted staging server).
|
||||
pub(crate) mod langfuse;
|
||||
|
||||
#[cfg(test)]
|
||||
mod journal_projection_tests;
|
||||
|
||||
/// Kind of run a trace belongs to, rendered as stable snake_case strings for
|
||||
/// Langfuse trace tags (`run:<type>`) and metadata (`run_type`) so runs can be
|
||||
/// filtered in the UI.
|
||||
|
||||
@@ -0,0 +1,336 @@
|
||||
//! Journal-backed span projection (C4 slice S2).
|
||||
//!
|
||||
//! Reconstructs a run's [`AgentProgress`] stream from the durable
|
||||
//! [`AgentObservation`] journal (the crate `AgentEvent` record) and folds it
|
||||
//! through the existing [`SpanCollector`], so trace spans no longer require the
|
||||
//! *live* in-run `AgentProgress` side-observer
|
||||
//! (`channels/providers/web/progress_bridge.rs`). A UI/supervisor can attach
|
||||
//! after a run, read the journal, and rebuild identical spans.
|
||||
//!
|
||||
//! This is deliberately built on `SpanCollector` (not a re-derivation) so span
|
||||
//! *shape* parity holds by construction for every `AgentProgress` the journal
|
||||
//! can produce. The one-way mapping here mirrors `OpenhumanEventBridge`
|
||||
//! (`tinyagents/observability.rs`) but is **pure** — it depends only on the
|
||||
//! journalled event, made possible by the crate carrying tool outcome
|
||||
//! (`duration_ms`/`output_bytes`/`error`) on `ToolCompleted` (tinyagents#18).
|
||||
//!
|
||||
//! Known parity gaps (see `docs/.../C4-journal-progress-parity-plan.md` §2a):
|
||||
//! - `ModelCallCompleted.cost_usd` and `cache_creation_tokens` are not on the
|
||||
//! crate event; filled as `0` here and to be sourced from the persisted
|
||||
//! per-run cost store at export time.
|
||||
//! - Sub-agent prompt/output content is not on the crate lifecycle events, so
|
||||
//! subagent spans carry lifecycle/timing and child tool/model structure but
|
||||
//! empty delegated prompt/final output until a richer journal event exists.
|
||||
|
||||
use tinyagents::harness::events::AgentEvent;
|
||||
use tinyagents::harness::observability::AgentObservation;
|
||||
|
||||
use super::{SpanCollector, TraceContext, TraceSpan};
|
||||
use crate::openhuman::agent::progress::AgentProgress;
|
||||
use crate::openhuman::tool_status::classify;
|
||||
|
||||
/// Mutable state threaded across a single run's observations while replaying.
|
||||
#[derive(Default)]
|
||||
struct ReplayState {
|
||||
/// 1-based iteration index, bumped once per `ModelStarted` — the same
|
||||
/// attribution the live `IterationCursor` provides.
|
||||
iteration: u32,
|
||||
/// Max iterations configured for the turn (carried onto iteration spans).
|
||||
max_iterations: u32,
|
||||
/// `call_id → model name`, learned from `ModelStarted` so the matching
|
||||
/// `ModelCompleted` can name its generation span (the crate `ModelCompleted`
|
||||
/// event carries no model name).
|
||||
models: std::collections::HashMap<String, String>,
|
||||
/// Stack of currently-open sub-agent runs. The crate lifecycle event only
|
||||
/// carries name/depth; ordered replay brackets child model/tool events.
|
||||
subagents: Vec<ReplaySubagent>,
|
||||
/// Monotonic suffix to make repeated invocations of the same child name
|
||||
/// distinct in the span tree.
|
||||
next_subagent_seq: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct ReplaySubagent {
|
||||
agent_id: String,
|
||||
task_id: String,
|
||||
depth: usize,
|
||||
iteration: u32,
|
||||
started_ts_ms: u64,
|
||||
}
|
||||
|
||||
impl ReplayState {
|
||||
fn active_subagent(&self) -> Option<&ReplaySubagent> {
|
||||
self.subagents.last()
|
||||
}
|
||||
|
||||
fn active_subagent_mut(&mut self) -> Option<&mut ReplaySubagent> {
|
||||
self.subagents.last_mut()
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps one journalled observation to zero or more [`AgentProgress`] events,
|
||||
/// updating `state`. Non-span-bearing events (deltas, budget/cache/steering
|
||||
/// diagnostics) map to nothing — `SpanCollector` ignores them anyway.
|
||||
fn observation_to_progress(obs: &AgentObservation, state: &mut ReplayState) -> Vec<AgentProgress> {
|
||||
let event = &obs.event;
|
||||
match event {
|
||||
AgentEvent::RunStarted { .. } => {
|
||||
if state.active_subagent().is_some() {
|
||||
Vec::new()
|
||||
} else {
|
||||
vec![AgentProgress::TurnStarted]
|
||||
}
|
||||
}
|
||||
|
||||
AgentEvent::ModelStarted { call_id, model } => {
|
||||
let iteration = match state.active_subagent_mut() {
|
||||
Some(scope) => {
|
||||
scope.iteration += 1;
|
||||
scope.iteration
|
||||
}
|
||||
None => {
|
||||
state.iteration += 1;
|
||||
state.iteration
|
||||
}
|
||||
};
|
||||
state
|
||||
.models
|
||||
.insert(call_id.as_str().to_string(), model.clone());
|
||||
match state.active_subagent() {
|
||||
Some(scope) => vec![AgentProgress::SubagentIterationStarted {
|
||||
agent_id: scope.agent_id.clone(),
|
||||
task_id: scope.task_id.clone(),
|
||||
iteration,
|
||||
max_iterations: state.max_iterations,
|
||||
extended_policy: false,
|
||||
}],
|
||||
None => vec![AgentProgress::IterationStarted {
|
||||
iteration,
|
||||
max_iterations: state.max_iterations,
|
||||
}],
|
||||
}
|
||||
}
|
||||
|
||||
AgentEvent::ToolStarted { call_id, tool_name } => match state.active_subagent() {
|
||||
Some(scope) => vec![AgentProgress::SubagentToolCallStarted {
|
||||
agent_id: scope.agent_id.clone(),
|
||||
task_id: scope.task_id.clone(),
|
||||
call_id: call_id.as_str().to_string(),
|
||||
tool_name: tool_name.clone(),
|
||||
arguments: serde_json::Value::Null,
|
||||
iteration: scope.iteration,
|
||||
display_label: None,
|
||||
display_detail: None,
|
||||
}],
|
||||
None => vec![AgentProgress::ToolCallStarted {
|
||||
call_id: call_id.as_str().to_string(),
|
||||
tool_name: tool_name.clone(),
|
||||
// The journal does not carry the model's raw argument JSON in
|
||||
// payload-free mode; the tool span still renders from name + id.
|
||||
arguments: serde_json::Value::Null,
|
||||
iteration: state.iteration,
|
||||
display_label: None,
|
||||
display_detail: None,
|
||||
}],
|
||||
},
|
||||
|
||||
AgentEvent::ToolCompleted {
|
||||
call_id,
|
||||
tool_name,
|
||||
input,
|
||||
output,
|
||||
duration_ms,
|
||||
output_bytes,
|
||||
error,
|
||||
..
|
||||
} => {
|
||||
// Outcome now rides the event (tinyagents#18): success, duration and
|
||||
// size are self-describing, and the same `classify` the live path
|
||||
// uses reproduces the identical `ClassifiedFailure` from the
|
||||
// journalled error string. `output` is present only when the run
|
||||
// captured payloads (full-content journals).
|
||||
let failure = error.as_ref().map(|text| classify(text, false));
|
||||
let output_text = match output {
|
||||
Some(serde_json::Value::String(text)) => text.clone(),
|
||||
Some(value) => value.to_string(),
|
||||
None => String::new(),
|
||||
};
|
||||
match state.active_subagent() {
|
||||
Some(scope) => vec![AgentProgress::SubagentToolCallCompleted {
|
||||
agent_id: scope.agent_id.clone(),
|
||||
task_id: scope.task_id.clone(),
|
||||
call_id: call_id.as_str().to_string(),
|
||||
tool_name: tool_name.clone(),
|
||||
success: error.is_none(),
|
||||
output_chars: output_bytes.unwrap_or(0) as usize,
|
||||
output: output_text,
|
||||
arguments: input.clone(),
|
||||
elapsed_ms: duration_ms.unwrap_or(0),
|
||||
iteration: scope.iteration,
|
||||
failure,
|
||||
}],
|
||||
None => vec![AgentProgress::ToolCallCompleted {
|
||||
call_id: call_id.as_str().to_string(),
|
||||
tool_name: tool_name.clone(),
|
||||
success: error.is_none(),
|
||||
output_chars: output_bytes.unwrap_or(0) as usize,
|
||||
output: output_text,
|
||||
arguments: input.clone(),
|
||||
elapsed_ms: duration_ms.unwrap_or(0),
|
||||
iteration: state.iteration,
|
||||
failure,
|
||||
}],
|
||||
}
|
||||
}
|
||||
|
||||
AgentEvent::ModelCompleted {
|
||||
call_id,
|
||||
usage,
|
||||
input,
|
||||
output,
|
||||
..
|
||||
} => {
|
||||
let model = state
|
||||
.models
|
||||
.get(call_id.as_str())
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
let usage = usage.unwrap_or_default();
|
||||
let scope = state.active_subagent().cloned();
|
||||
let iteration = scope
|
||||
.as_ref()
|
||||
.map(|s| s.iteration)
|
||||
.unwrap_or(state.iteration);
|
||||
let mut progress = vec![AgentProgress::ModelCallCompleted {
|
||||
model,
|
||||
// Provider qualification/cost are filled at export time from the
|
||||
// persisted cost store, not the journal (§2a).
|
||||
provider_id: String::new(),
|
||||
subagent_task_id: scope.as_ref().map(|s| s.task_id.clone()),
|
||||
input: input.clone(),
|
||||
output: output.clone(),
|
||||
iteration,
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
cached_input_tokens: usage.cache_read_tokens,
|
||||
cache_creation_tokens: 0,
|
||||
reasoning_tokens: usage.reasoning_tokens,
|
||||
cost_usd: 0.0,
|
||||
}];
|
||||
if scope.is_none() {
|
||||
let turn_input = input.as_ref().map(json_content_text);
|
||||
let turn_output = output.as_ref().map(json_content_text);
|
||||
if turn_input.is_some() || turn_output.is_some() {
|
||||
progress.push(AgentProgress::TurnContent {
|
||||
input: turn_input,
|
||||
output: turn_output,
|
||||
});
|
||||
}
|
||||
}
|
||||
progress
|
||||
}
|
||||
|
||||
AgentEvent::SubAgentStarted { name, depth } => {
|
||||
state.next_subagent_seq += 1;
|
||||
let task_id = format!("{name}-d{depth}-{}", state.next_subagent_seq);
|
||||
state.subagents.push(ReplaySubagent {
|
||||
agent_id: name.clone(),
|
||||
task_id: task_id.clone(),
|
||||
depth: *depth,
|
||||
iteration: 0,
|
||||
started_ts_ms: obs.ts_ms,
|
||||
});
|
||||
vec![AgentProgress::SubagentSpawned {
|
||||
agent_id: name.clone(),
|
||||
task_id,
|
||||
mode: "typed".to_string(),
|
||||
dedicated_thread: false,
|
||||
prompt_chars: 0,
|
||||
worker_thread_id: None,
|
||||
display_name: Some(name.clone()),
|
||||
prompt: String::new(),
|
||||
}]
|
||||
}
|
||||
|
||||
AgentEvent::SubAgentCompleted { name, depth } => {
|
||||
let pos = state
|
||||
.subagents
|
||||
.iter()
|
||||
.rposition(|scope| scope.agent_id == *name && scope.depth == *depth);
|
||||
let Some(scope) = pos.map(|index| state.subagents.remove(index)) else {
|
||||
return Vec::new();
|
||||
};
|
||||
vec![AgentProgress::SubagentCompleted {
|
||||
agent_id: scope.agent_id,
|
||||
task_id: scope.task_id,
|
||||
elapsed_ms: obs.ts_ms.saturating_sub(scope.started_ts_ms),
|
||||
iterations: scope.iteration,
|
||||
output_chars: 0,
|
||||
output: String::new(),
|
||||
worktree_path: None,
|
||||
changed_files: Vec::new(),
|
||||
dirty_status: None,
|
||||
}]
|
||||
}
|
||||
|
||||
AgentEvent::RunCompleted { .. } => {
|
||||
if state.active_subagent().is_some() {
|
||||
Vec::new()
|
||||
} else {
|
||||
vec![AgentProgress::TurnCompleted {
|
||||
iterations: state.iteration,
|
||||
}]
|
||||
}
|
||||
}
|
||||
|
||||
AgentEvent::RunFailed { error, .. } => {
|
||||
let Some(scope) = state.subagents.pop() else {
|
||||
return Vec::new();
|
||||
};
|
||||
vec![AgentProgress::SubagentFailed {
|
||||
agent_id: scope.agent_id,
|
||||
task_id: scope.task_id,
|
||||
error: error.clone(),
|
||||
}]
|
||||
}
|
||||
|
||||
_ => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn json_content_text(value: &serde_json::Value) -> String {
|
||||
match value {
|
||||
serde_json::Value::String(text) => text.clone(),
|
||||
serde_json::Value::Object(map) => map
|
||||
.get("content")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.map(str::to_string)
|
||||
.unwrap_or_else(|| value.to_string()),
|
||||
_ => value.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Projects a run's journalled `observations` into trace spans by replaying
|
||||
/// them into [`AgentProgress`] and folding through a fresh [`SpanCollector`],
|
||||
/// stamped with each observation's journal timestamp (`ts_ms`).
|
||||
pub(crate) fn spans_from_observations(
|
||||
ctx: TraceContext,
|
||||
max_iterations: u32,
|
||||
observations: &[AgentObservation],
|
||||
) -> Vec<TraceSpan> {
|
||||
let capture_content = ctx.capture_content;
|
||||
let mut collector = SpanCollector::new(ctx).with_content_capture(capture_content);
|
||||
let mut state = ReplayState {
|
||||
max_iterations,
|
||||
..ReplayState::default()
|
||||
};
|
||||
let mut last_ts = 0;
|
||||
for obs in observations {
|
||||
last_ts = obs.ts_ms;
|
||||
for progress in observation_to_progress(obs, &mut state) {
|
||||
collector.record(&progress, obs.ts_ms);
|
||||
}
|
||||
}
|
||||
collector.finish(last_ts);
|
||||
collector.spans().to_vec()
|
||||
}
|
||||
@@ -0,0 +1,334 @@
|
||||
use super::journal_projection::spans_from_observations;
|
||||
use super::{SpanKind, SpanStatus, TraceContext};
|
||||
use tinyagents::harness::events::AgentEvent;
|
||||
use tinyagents::harness::ids::{CallId, EventId, RunId};
|
||||
use tinyagents::harness::observability::AgentObservation;
|
||||
use tinyagents::harness::usage::Usage;
|
||||
|
||||
/// Wraps an event as a journalled observation stamped at `ts`.
|
||||
fn obs(offset: u64, ts: u64, event: AgentEvent) -> AgentObservation {
|
||||
AgentObservation {
|
||||
event_id: EventId::new(format!("run-1-evt-{offset}")),
|
||||
run_id: RunId::new("run-1"),
|
||||
parent_run_id: None,
|
||||
root_run_id: RunId::new("run-1"),
|
||||
offset,
|
||||
ts_ms: ts,
|
||||
event,
|
||||
}
|
||||
}
|
||||
|
||||
fn tool_completed(call: &str, name: &str, error: Option<&str>) -> AgentEvent {
|
||||
AgentEvent::ToolCompleted {
|
||||
call_id: CallId::new(call),
|
||||
tool_name: name.to_string(),
|
||||
started_at_ms: Some(1_020),
|
||||
input: None,
|
||||
output: None,
|
||||
duration_ms: Some(30),
|
||||
output_bytes: Some(12),
|
||||
error: error.map(str::to_string),
|
||||
}
|
||||
}
|
||||
|
||||
fn single_turn(tool_error: Option<&str>) -> Vec<AgentObservation> {
|
||||
vec![
|
||||
obs(
|
||||
0,
|
||||
1_000,
|
||||
AgentEvent::RunStarted {
|
||||
run_id: RunId::new("run-1"),
|
||||
thread_id: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
1,
|
||||
1_010,
|
||||
AgentEvent::ModelStarted {
|
||||
call_id: CallId::new("c1"),
|
||||
model: "gpt-4".to_string(),
|
||||
},
|
||||
),
|
||||
obs(
|
||||
2,
|
||||
1_020,
|
||||
AgentEvent::ToolStarted {
|
||||
call_id: CallId::new("t1"),
|
||||
tool_name: "lookup".to_string(),
|
||||
},
|
||||
),
|
||||
obs(3, 1_050, tool_completed("t1", "lookup", tool_error)),
|
||||
obs(
|
||||
4,
|
||||
1_060,
|
||||
AgentEvent::ModelCompleted {
|
||||
call_id: CallId::new("c1"),
|
||||
started_at_ms: Some(1_010),
|
||||
usage: Some(Usage::new(100, 20)),
|
||||
input: None,
|
||||
output: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
5,
|
||||
1_070,
|
||||
AgentEvent::RunCompleted {
|
||||
run_id: RunId::new("run-1"),
|
||||
},
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
fn subagent_turn() -> Vec<AgentObservation> {
|
||||
vec![
|
||||
obs(
|
||||
0,
|
||||
1_000,
|
||||
AgentEvent::RunStarted {
|
||||
run_id: RunId::new("run-1"),
|
||||
thread_id: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
1,
|
||||
1_010,
|
||||
AgentEvent::SubAgentStarted {
|
||||
name: "researcher".to_string(),
|
||||
depth: 1,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
2,
|
||||
1_020,
|
||||
AgentEvent::ModelStarted {
|
||||
call_id: CallId::new("scout-model"),
|
||||
model: "gpt-4".to_string(),
|
||||
},
|
||||
),
|
||||
obs(
|
||||
3,
|
||||
1_030,
|
||||
AgentEvent::ToolStarted {
|
||||
call_id: CallId::new("scout-tool"),
|
||||
tool_name: "read_file".to_string(),
|
||||
},
|
||||
),
|
||||
obs(4, 1_060, tool_completed("scout-tool", "read_file", None)),
|
||||
obs(
|
||||
5,
|
||||
1_070,
|
||||
AgentEvent::ModelCompleted {
|
||||
call_id: CallId::new("scout-model"),
|
||||
started_at_ms: Some(1_020),
|
||||
usage: Some(Usage::new(11, 7)),
|
||||
input: None,
|
||||
output: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
6,
|
||||
1_090,
|
||||
AgentEvent::SubAgentCompleted {
|
||||
name: "researcher".to_string(),
|
||||
depth: 1,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
7,
|
||||
1_100,
|
||||
AgentEvent::RunCompleted {
|
||||
run_id: RunId::new("run-1"),
|
||||
},
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
fn ctx() -> TraceContext {
|
||||
TraceContext::new("session-1", Some("user-1".into()))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_single_agent_turn_span_tree() {
|
||||
let spans = spans_from_observations(ctx(), 10, &single_turn(None));
|
||||
|
||||
// Turn + iteration + tool + generation spans are all present.
|
||||
let by_kind = |k: SpanKind| spans.iter().filter(|s| s.kind == k).count();
|
||||
assert_eq!(by_kind(SpanKind::Turn), 1, "one turn span");
|
||||
assert_eq!(by_kind(SpanKind::Iteration), 1, "one iteration span");
|
||||
assert_eq!(by_kind(SpanKind::Tool), 1, "one tool span");
|
||||
assert_eq!(by_kind(SpanKind::Generation), 1, "one generation span");
|
||||
|
||||
let tool = spans.iter().find(|s| s.kind == SpanKind::Tool).unwrap();
|
||||
assert_eq!(tool.name, "tool.lookup");
|
||||
assert_eq!(tool.attributes["tool.success"], serde_json::json!(true));
|
||||
assert_eq!(tool.attributes["tool.output_chars"], serde_json::json!(12));
|
||||
assert_eq!(tool.attributes["tool.elapsed_ms"], serde_json::json!(30));
|
||||
|
||||
let generation = spans
|
||||
.iter()
|
||||
.find(|s| s.kind == SpanKind::Generation)
|
||||
.unwrap();
|
||||
assert!(
|
||||
generation.name.contains("gpt-4"),
|
||||
"gen span names the model"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_tool_projects_error_outcome() {
|
||||
// With content capture on, a failed tool span carries the classified
|
||||
// cause reconstructed from the journalled error string, reproducing the
|
||||
// live path's `classify(error, false)` exactly.
|
||||
let spans = spans_from_observations(
|
||||
ctx().with_capture_content(true),
|
||||
10,
|
||||
&single_turn(Some("permission denied opening /etc/x")),
|
||||
);
|
||||
let tool = spans.iter().find(|s| s.kind == SpanKind::Tool).unwrap();
|
||||
assert_eq!(tool.attributes["tool.success"], serde_json::json!(false));
|
||||
assert!(
|
||||
tool.attributes.contains_key("error.message"),
|
||||
"failed tool span carries a classified error message"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_subagent_scope_from_lifecycle_brackets() {
|
||||
let spans = spans_from_observations(ctx(), 10, &subagent_turn());
|
||||
|
||||
let by_kind = |k: SpanKind| spans.iter().filter(|s| s.kind == k).count();
|
||||
assert_eq!(by_kind(SpanKind::Subagent), 1, "one subagent span");
|
||||
assert_eq!(
|
||||
by_kind(SpanKind::SubagentIteration),
|
||||
1,
|
||||
"one child iteration span"
|
||||
);
|
||||
assert_eq!(by_kind(SpanKind::Tool), 1, "one child tool span");
|
||||
assert_eq!(by_kind(SpanKind::Generation), 1, "one child generation");
|
||||
|
||||
let subagent = spans.iter().find(|s| s.kind == SpanKind::Subagent).unwrap();
|
||||
assert_eq!(
|
||||
subagent.attributes["subagent.agent_id"],
|
||||
serde_json::json!("researcher")
|
||||
);
|
||||
assert_eq!(
|
||||
subagent.attributes["subagent.iterations"],
|
||||
serde_json::json!(1)
|
||||
);
|
||||
|
||||
let child_iteration = spans
|
||||
.iter()
|
||||
.find(|s| s.kind == SpanKind::SubagentIteration)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
child_iteration.parent_span_id.as_deref(),
|
||||
Some(subagent.span_id.as_str())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_failed_subagent_from_child_run_failed() {
|
||||
let observations = vec![
|
||||
obs(
|
||||
0,
|
||||
1_000,
|
||||
AgentEvent::RunStarted {
|
||||
run_id: RunId::new("run-1"),
|
||||
thread_id: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
1,
|
||||
1_010,
|
||||
AgentEvent::SubAgentStarted {
|
||||
name: "researcher".to_string(),
|
||||
depth: 1,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
2,
|
||||
1_020,
|
||||
AgentEvent::RunFailed {
|
||||
run_id: RunId::new("run-1"),
|
||||
error: "provider unavailable".to_string(),
|
||||
},
|
||||
),
|
||||
obs(
|
||||
3,
|
||||
1_030,
|
||||
AgentEvent::RunCompleted {
|
||||
run_id: RunId::new("run-1"),
|
||||
},
|
||||
),
|
||||
];
|
||||
|
||||
let spans = spans_from_observations(ctx(), 10, &observations);
|
||||
let subagent = spans.iter().find(|s| s.kind == SpanKind::Subagent).unwrap();
|
||||
|
||||
assert_eq!(subagent.status, SpanStatus::Error);
|
||||
assert_eq!(subagent.attributes["error"], serde_json::json!(true));
|
||||
assert!(
|
||||
subagent.attributes.get("error.length").is_some(),
|
||||
"failed subagent span carries redacted error metadata"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_turn_content_from_root_model_io() {
|
||||
let observations = vec![
|
||||
obs(
|
||||
0,
|
||||
1_000,
|
||||
AgentEvent::RunStarted {
|
||||
run_id: RunId::new("run-1"),
|
||||
thread_id: None,
|
||||
},
|
||||
),
|
||||
obs(
|
||||
1,
|
||||
1_010,
|
||||
AgentEvent::ModelStarted {
|
||||
call_id: CallId::new("m1"),
|
||||
model: "gpt-4".to_string(),
|
||||
},
|
||||
),
|
||||
obs(
|
||||
2,
|
||||
1_020,
|
||||
AgentEvent::ModelCompleted {
|
||||
call_id: CallId::new("m1"),
|
||||
started_at_ms: Some(1_010),
|
||||
usage: Some(Usage::new(5, 3)),
|
||||
input: Some(serde_json::json!([
|
||||
{"role": "user", "content": "summarize this"}
|
||||
])),
|
||||
output: Some(serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": "short summary"
|
||||
})),
|
||||
},
|
||||
),
|
||||
obs(
|
||||
3,
|
||||
1_030,
|
||||
AgentEvent::RunCompleted {
|
||||
run_id: RunId::new("run-1"),
|
||||
},
|
||||
),
|
||||
];
|
||||
let spans = spans_from_observations(ctx().with_capture_content(true), 10, &observations);
|
||||
let turn = spans.iter().find(|s| s.kind == SpanKind::Turn).unwrap();
|
||||
|
||||
assert!(
|
||||
turn.input
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.to_string()
|
||||
.contains("summarize this"),
|
||||
"root model input is attached through TurnContent"
|
||||
);
|
||||
assert_eq!(
|
||||
turn.output.as_ref().unwrap(),
|
||||
&serde_json::json!("short summary")
|
||||
);
|
||||
}
|
||||
@@ -27,6 +27,10 @@ pub enum AgentTurnOrigin {
|
||||
WebChat {
|
||||
thread_id: String,
|
||||
client_id: String,
|
||||
/// Per-turn request id, when the caller has one. Used by internal
|
||||
/// observers to correlate a live progress bridge with the durable
|
||||
/// tinyagents journal stream for the same turn.
|
||||
request_id: Option<String>,
|
||||
},
|
||||
/// Inbound message from a non-web channel (Telegram / Discord / Slack /
|
||||
/// Yuanbao / etc.). External-effect tools must persist a
|
||||
@@ -161,6 +165,7 @@ mod tests {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "outer".into(),
|
||||
client_id: "c-outer".into(),
|
||||
request_id: Some("req-outer".into()),
|
||||
},
|
||||
async {
|
||||
with_origin(
|
||||
|
||||
@@ -912,6 +912,7 @@ mod tests {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "t-test".into(),
|
||||
client_id: "c-test".into(),
|
||||
request_id: Some("req-test".into()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1248,6 +1249,7 @@ mod tests {
|
||||
let origin = AgentTurnOrigin::WebChat {
|
||||
thread_id: "thread-42".into(),
|
||||
client_id: "client-1".into(),
|
||||
request_id: Some("req-42".into()),
|
||||
};
|
||||
let handle = tokio::spawn(async move {
|
||||
turn_origin::with_origin(
|
||||
|
||||
@@ -550,6 +550,7 @@ pub async fn start_chat(
|
||||
let origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::WebChat {
|
||||
thread_id: thread_id_task.clone(),
|
||||
client_id: client_id_task.clone(),
|
||||
request_id: Some(request_id_task.clone()),
|
||||
};
|
||||
// `None` => the turn was cancelled cooperatively before producing a
|
||||
// result; the interrupting/cancelling side already emitted the
|
||||
@@ -780,6 +781,7 @@ async fn spawn_parallel_turn(
|
||||
let origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::WebChat {
|
||||
thread_id: thread_id_task.clone(),
|
||||
client_id: client_id_task.clone(),
|
||||
request_id: Some(request_id_task.clone()),
|
||||
};
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
|
||||
@@ -129,6 +129,97 @@ fn session_profile_user_attribution(config: &crate::openhuman::config::Config) -
|
||||
.or(state.user_id)
|
||||
}
|
||||
|
||||
fn span_projection_signature(
|
||||
spans: &[crate::openhuman::agent::progress_tracing::TraceSpan],
|
||||
) -> Vec<String> {
|
||||
spans
|
||||
.iter()
|
||||
.map(|span| {
|
||||
let attr_keys = span
|
||||
.attributes
|
||||
.keys()
|
||||
.map(String::as_str)
|
||||
.collect::<Vec<_>>()
|
||||
.join(",");
|
||||
format!(
|
||||
"{:?}|{}|{:?}|attrs:[{}]",
|
||||
span.kind, span.name, span.status, attr_keys
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn shadow_compare_journal_projection(
|
||||
request_id: &str,
|
||||
trace_ctx: crate::openhuman::agent::progress_tracing::TraceContext,
|
||||
max_iterations: u32,
|
||||
live_spans: &[crate::openhuman::agent::progress_tracing::TraceSpan],
|
||||
) {
|
||||
let Some(journal_run_id) =
|
||||
crate::openhuman::tinyagents::journal::take_request_journal_run(request_id)
|
||||
else {
|
||||
log::debug!(
|
||||
"[agent-tracing][journal-shadow] no journal run registered request_id={}",
|
||||
request_id
|
||||
);
|
||||
return;
|
||||
};
|
||||
|
||||
let observations = match crate::openhuman::tinyagents::journal::read_run_events(
|
||||
&journal_run_id,
|
||||
0,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(observations) => observations,
|
||||
Err(err) => {
|
||||
log::warn!(
|
||||
"[agent-tracing][journal-shadow] read failed request_id={} journal_run_id={} err={err}",
|
||||
request_id,
|
||||
journal_run_id
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
if observations.is_empty() {
|
||||
log::warn!(
|
||||
"[agent-tracing][journal-shadow] journal empty request_id={} journal_run_id={}",
|
||||
request_id,
|
||||
journal_run_id
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let projected =
|
||||
crate::openhuman::agent::progress_tracing::journal_projection::spans_from_observations(
|
||||
trace_ctx,
|
||||
max_iterations,
|
||||
&observations,
|
||||
);
|
||||
let live_sig = span_projection_signature(live_spans);
|
||||
let projected_sig = span_projection_signature(&projected);
|
||||
if live_sig == projected_sig {
|
||||
log::debug!(
|
||||
"[agent-tracing][journal-shadow] parity ok request_id={} journal_run_id={} spans={} observations={}",
|
||||
request_id,
|
||||
journal_run_id,
|
||||
live_spans.len(),
|
||||
observations.len()
|
||||
);
|
||||
} else {
|
||||
log::warn!(
|
||||
"[agent-tracing][journal-shadow] parity divergence request_id={} journal_run_id={} live_spans={} journal_spans={} observations={} live_sig={:?} journal_sig={:?}",
|
||||
request_id,
|
||||
journal_run_id,
|
||||
live_spans.len(),
|
||||
projected.len(),
|
||||
observations.len(),
|
||||
live_sig,
|
||||
projected_sig
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn a background task that reads [`AgentProgress`] events from the
|
||||
/// agent turn loop and translates them into [`WebChannelEvent`]s tagged
|
||||
/// with the correct client/thread/request IDs. The task runs until the
|
||||
@@ -159,6 +250,7 @@ pub(crate) fn spawn_progress_bridge(
|
||||
metadata.session_id,
|
||||
);
|
||||
let mut round: u32 = 0;
|
||||
let mut parent_max_iterations: u32 = 0;
|
||||
let mut events_seen: u64 = 0;
|
||||
let mut parent_completed = false;
|
||||
let mut parent_tool_count: u64 = 0;
|
||||
@@ -170,6 +262,7 @@ pub(crate) fn spawn_progress_bridge(
|
||||
// progress stream into OTel/Langfuse-style spans correlated by session
|
||||
// id (falling back to the thread id for headless/autonomous runs).
|
||||
// `None` (disabled) is zero-cost.
|
||||
let mut journal_trace_ctx = None;
|
||||
let mut span_collector = if config.observability.share_usage_data
|
||||
|| config.observability.agent_tracing.enabled
|
||||
{
|
||||
@@ -226,6 +319,7 @@ pub(crate) fn spawn_progress_bridge(
|
||||
if let Some(agent_id) = metadata.agent_id.clone() {
|
||||
trace_ctx = trace_ctx.with_agent_id(agent_id);
|
||||
}
|
||||
journal_trace_ctx = Some(trace_ctx.clone());
|
||||
Some(SpanCollector::new(trace_ctx))
|
||||
} else {
|
||||
None
|
||||
@@ -411,6 +505,7 @@ pub(crate) fn spawn_progress_bridge(
|
||||
max_iterations,
|
||||
} => {
|
||||
round = iteration;
|
||||
parent_max_iterations = max_iterations;
|
||||
publish_web_channel_event(WebChannelEvent {
|
||||
event: "iteration_start".to_string(),
|
||||
client_id: client_id.clone(),
|
||||
@@ -1220,8 +1315,17 @@ pub(crate) fn spawn_progress_bridge(
|
||||
// never affects the turn outcome.
|
||||
if let Some(mut collector) = span_collector.take() {
|
||||
collector.finish(unix_epoch_ms());
|
||||
crate::openhuman::agent::progress_tracing::export_run_trace(&config, collector.spans())
|
||||
let live_spans = collector.spans().to_vec();
|
||||
if let Some(trace_ctx) = journal_trace_ctx.take() {
|
||||
shadow_compare_journal_projection(
|
||||
&request_id,
|
||||
trace_ctx,
|
||||
parent_max_iterations,
|
||||
&live_spans,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
crate::openhuman::agent::progress_tracing::export_run_trace(&config, &live_spans).await;
|
||||
}
|
||||
|
||||
log::debug!(
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
//! that don't need it.
|
||||
|
||||
use super::*;
|
||||
use crate::openhuman::memory_store::chunks::types::chunk_id;
|
||||
use crate::openhuman::memory_store::chunks::types::{chunk_id, Metadata, SourceRef};
|
||||
use chrono::TimeZone;
|
||||
use rusqlite::params;
|
||||
use tempfile::TempDir;
|
||||
|
||||
@@ -183,6 +183,7 @@ mod tests {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "t-int".into(),
|
||||
client_id: "c-int".into(),
|
||||
request_id: Some("req-int".into()),
|
||||
},
|
||||
tool.execute(json!({ "summary": "plan", "steps": ["one"] })),
|
||||
);
|
||||
|
||||
@@ -45,10 +45,12 @@
|
||||
//!
|
||||
//! [`EventSink::with_stream_id`]: tinyagents::harness::events::EventSink::with_stream_id
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
use tinyagents::error::Result as TaResult;
|
||||
use tinyagents::harness::events::{EventSink, HarnessRunStatus};
|
||||
@@ -66,6 +68,12 @@ use crate::openhuman::session_import::ops::open_session_stores;
|
||||
/// it round-trips the crate [`FileStore`] name sanitizer.
|
||||
const STATUS_NS: &str = "run_status";
|
||||
|
||||
/// Best-effort live request → durable tinyagents journal stream map. The web
|
||||
/// progress bridge uses this at turn end to shadow-project spans from the
|
||||
/// journal and compare them against the live `SpanCollector` path during C4 S3.
|
||||
static REQUEST_JOURNAL_RUNS: Lazy<Mutex<HashMap<String, String>>> =
|
||||
Lazy::new(|| Mutex::new(HashMap::new()));
|
||||
|
||||
/// Mints a fresh, slash-free, process-unique run id (`run.<32-hex>`), used three
|
||||
/// ways for one turn: the [`EventSink::with_stream_id`] prefix (so persisted
|
||||
/// `event_id`s are the restart-stable `{run_id}-evt-{offset}`), the journal
|
||||
@@ -81,6 +89,32 @@ pub(crate) fn mint_run_id() -> RunId {
|
||||
RunId::new(format!("run.{}", uuid::Uuid::new_v4().simple()))
|
||||
}
|
||||
|
||||
pub(crate) fn register_request_journal_run(request_id: &str, run_id: &str) {
|
||||
match REQUEST_JOURNAL_RUNS.lock() {
|
||||
Ok(mut runs) => {
|
||||
runs.insert(request_id.to_string(), run_id.to_string());
|
||||
log::debug!(
|
||||
"[journal] registered request journal request_id={} run_id={}",
|
||||
request_id,
|
||||
run_id
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
log::debug!(
|
||||
"[journal] request journal registry poisoned request_id={} err={err}",
|
||||
request_id
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn take_request_journal_run(request_id: &str) -> Option<String> {
|
||||
REQUEST_JOURNAL_RUNS
|
||||
.lock()
|
||||
.ok()
|
||||
.and_then(|mut runs| runs.remove(request_id))
|
||||
}
|
||||
|
||||
/// Resolve the internal workspace directory (`{workspace}`) whose
|
||||
/// `tinyagents_store/` subtree holds the journal + kv stores. Async because the
|
||||
/// config load is async; errors are surfaced so callers can log-and-skip.
|
||||
|
||||
@@ -697,6 +697,15 @@ pub(crate) async fn run_turn_via_tinyagents_shared(
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
if subagent_scope.is_none() {
|
||||
if let Some(crate::openhuman::agent::turn_origin::AgentTurnOrigin::WebChat {
|
||||
request_id: Some(request_id),
|
||||
..
|
||||
}) = crate::openhuman::agent::turn_origin::current()
|
||||
{
|
||||
journal::register_request_journal_run(&request_id, journal_run_id.as_str());
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(events) = &events {
|
||||
ctx = ctx.with_events(events.clone());
|
||||
|
||||
@@ -916,11 +916,11 @@ impl EventListener for OpenhumanEventBridge {
|
||||
tool_name,
|
||||
input,
|
||||
output,
|
||||
// tinyagents 1.7 added started_at_ms/duration_ms/output_bytes/error
|
||||
// to this event. The bridge keeps sourcing success/duration/size
|
||||
// from its own capture side channel (failure_map/tool_started_at)
|
||||
// below, so the new crate-provided fields are intentionally ignored
|
||||
// here to preserve existing behavior. TODO: adopt them directly.
|
||||
// `started_at_ms`/`duration_ms`/`output_bytes`/`error` now ride
|
||||
// the crate event (tinyagents 1.7 / tinyagents#18). The bridge
|
||||
// still reads its richer side channels below to preserve current
|
||||
// success/duration/size behavior; adopting crate fields directly
|
||||
// is C4 slice S1.
|
||||
..
|
||||
} => {
|
||||
let iteration = self.iteration();
|
||||
|
||||
@@ -12,7 +12,8 @@ use async_trait::async_trait;
|
||||
use tinyagents::harness::steering::{SteeringCommand, SteeringHandle};
|
||||
use tinyagents::harness::tool::{
|
||||
SandboxMode, Tool, ToolAccess, ToolCall as TaToolCall, ToolExecutionContext, ToolPolicy,
|
||||
ToolResult as TaToolResult, ToolRuntime, ToolSchema, ToolSideEffects, WorkspaceAccess,
|
||||
ToolResult as TaToolResult, ToolRuntime, ToolSchema, ToolSideEffects,
|
||||
ToolTimeout as TaToolTimeout, WorkspaceAccess,
|
||||
};
|
||||
|
||||
/// A captured early-exit: a sub-agent invoked an early-exit tool (e.g.
|
||||
@@ -153,12 +154,10 @@ pub(crate) fn tool_policy_from_openhuman_tool(
|
||||
})
|
||||
.with_runtime(ToolRuntime {
|
||||
timeout_ms,
|
||||
// tinyagents 1.7 added a structured `timeout: ToolTimeout` field
|
||||
// alongside the numeric `timeout_ms`. Inherit the run/global policy
|
||||
// to preserve prior behavior (the numeric `timeout_ms` above stays
|
||||
// the only per-tool deadline signal). Fully-qualified because the
|
||||
// local `ToolTimeout` import here is openhuman's, not the harness's.
|
||||
timeout: tinyagents::harness::tool::ToolTimeout::Inherit,
|
||||
// tinyagents 1.7 added a structured timeout field alongside the
|
||||
// numeric timeout_ms. Inherit the run/global policy and keep the
|
||||
// numeric timeout_ms above as the only per-tool deadline signal.
|
||||
timeout: TaToolTimeout::Inherit,
|
||||
max_retries: None,
|
||||
idempotent: tool.is_concurrency_safe(&serde_json::Value::Null),
|
||||
cancelable: true,
|
||||
|
||||
@@ -1185,6 +1185,7 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "approval-raw-thread".to_string(),
|
||||
client_id: "approval-raw-client".to_string(),
|
||||
request_id: None,
|
||||
},
|
||||
APPROVAL_CHAT_CONTEXT.scope(
|
||||
ApprovalChatContext {
|
||||
@@ -1373,6 +1374,7 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "approval-auto-thread".to_string(),
|
||||
client_id: "approval-auto-client".to_string(),
|
||||
request_id: None,
|
||||
},
|
||||
gate.intercept_audited(
|
||||
"tools.composio_execute",
|
||||
@@ -1424,6 +1426,7 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "approval-deny-thread".to_string(),
|
||||
client_id: "approval-deny-client".to_string(),
|
||||
request_id: None,
|
||||
},
|
||||
APPROVAL_CHAT_CONTEXT.scope(
|
||||
ApprovalChatContext {
|
||||
@@ -1542,6 +1545,7 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() {
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "approval-persist-failure-thread".to_string(),
|
||||
client_id: "approval-persist-failure-client".to_string(),
|
||||
request_id: None,
|
||||
},
|
||||
APPROVAL_CHAT_CONTEXT.scope(
|
||||
ApprovalChatContext {
|
||||
|
||||
@@ -592,6 +592,7 @@ async fn approval_gate_rpc_decision_resumes_parked_tool_and_records_execution()
|
||||
AgentTurnOrigin::WebChat {
|
||||
thread_id: "worker-b-thread".to_string(),
|
||||
client_id: "worker-b-client".to_string(),
|
||||
request_id: None,
|
||||
},
|
||||
APPROVAL_CHAT_CONTEXT.scope(
|
||||
ApprovalChatContext {
|
||||
|
||||
Reference in New Issue
Block a user