diff --git a/cmd/odek/subagent.go b/cmd/odek/subagent.go index 051babe..e395b77 100644 --- a/cmd/odek/subagent.go +++ b/cmd/odek/subagent.go @@ -318,13 +318,46 @@ type subagentTelemetryWriter struct { taskID string step int mu sync.Mutex + wire subagentWireContext // P1/P3 decorations; zero value = legacy records } func newSubagentTelemetryWriter(w io.Writer, taskID string) *subagentTelemetryWriter { + return newSubagentTelemetryWriterWithWire(w, taskID, subagentWireContext{}) +} + +// newSubagentTelemetryWriterWithWire attaches the child's post-resolution +// wire context (P1 profile/risk identity, P3 budget block, cost estimator +// and the engine usage probe) to the lifecycle records. +func newSubagentTelemetryWriterWithWire(w io.Writer, taskID string, wire subagentWireContext) *subagentTelemetryWriter { if w == nil || taskID == "" { return nil } - return &subagentTelemetryWriter{w: w, taskID: taskID} + return &subagentTelemetryWriter{w: w, taskID: taskID, wire: wire} +} + +// emitStarted reports the lifecycle start record: the pre-existing +// pid/depth/timeout/max_iter fields plus the P1 identity fields (resolved +// profile id, effective post-clamp risk cap — each omitted when empty) and +// the P3 budget block. cost_usd is deliberately absent: nothing has been +// spent at start. +func (t *subagentTelemetryWriter) emitStarted(pid, depth, timeoutSeconds, maxIterations int) { + rec := map[string]any{ + "type": "subagent_started", + "pid": pid, + "depth": depth, + "timeout_s": timeoutSeconds, + "max_iter": maxIterations, + } + if t.wire.Profile != "" { + rec["profile"] = t.wire.Profile + } + if t.wire.MaxRisk != "" { + rec["max_risk"] = t.wire.MaxRisk + } + for k, v := range t.wire.Budget.fields() { + rec[k] = v + } + t.emit(rec) } // emit writes one compact NDJSON record with the task_id echoed. @@ -343,16 +376,161 @@ func (t *subagentTelemetryWriter) emit(record map[string]any) { // emitProgress reports a tool-start step. The tool NAME is included; // arguments and outputs never are — they are model-controlled content and // the telemetry path must stay argument-free (telemetry plan, security §4). +// P3: each record also carries the budget block and the cumulative cost +// estimate (cost_usd) so a client can render %-of-budget per step without +// remembering the started record. func (t *subagentTelemetryWriter) emitProgress(tool string) { t.mu.Lock() t.step++ n := t.step t.mu.Unlock() - t.emit(map[string]any{ + rec := map[string]any{ "type": "subagent_progress", "step": n, "tool": tool, - }) + } + for k, v := range t.wire.Budget.fields() { + rec[k] = v + } + if t.wire.Usage != nil && t.wire.Cost.configured() { + in, out := t.wire.Usage() + rec["cost_usd"] = t.wire.Cost.estimate(in, out) + } + t.emit(rec) +} + +// emitFinished reports the terminal lifecycle record with the final +// estimated cost (P6). The cost rides the same /api/usage math over the +// engine's provider-reported token totals and is omitted when no price +// side is configured — the wire never emits a fabricated $0. +func (t *subagentTelemetryWriter) emitFinished(status string, iterations int, durationSeconds float64, tokensUsed int) { + rec := map[string]any{ + "type": "subagent_finished", + "status": status, + "iterations": iterations, + "duration_s": durationSeconds, + "tokens_used": tokensUsed, + } + if t.wire.Usage != nil && t.wire.Cost.configured() { + in, out := t.wire.Usage() + rec["cost_usd"] = t.wire.Cost.estimate(in, out) + } + t.emit(rec) +} + +// ── Wire additions (P1/P3/P4/P6 — child half) ──────────────────────── + +// subagentWireBudget is the child's effective post-resolution budget block +// (P3): the wall-clock budget, the iteration budget, and the enforced cost +// cap. It mirrors the three headline numbers the lifespan block +// (buildLifespanBlock) announces to the child itself, so parents and UIs +// render progress against exactly the budgets the child enforces. +// CostUSD is stamped only when budget cost enforcement is active +// (Limits.CostEnforcementActive) — a cap without resolved prices is not an +// enforced cap, and the wire never reports $0. +type subagentWireBudget struct { + Seconds int + Iterations int + CostUSD float64 +} + +func newSubagentWireBudget(limits budget.Limits, timeoutSeconds, maxIterations int) subagentWireBudget { + b := subagentWireBudget{Seconds: timeoutSeconds, Iterations: maxIterations} + if limits.CostEnforcementActive() { + b.CostUSD = limits.MaxCostUSD + } + return b +} + +// fields renders the non-zero entries for a telemetry record. Zero values +// are omitted: an absent field means "no cap configured on this wire +// version" (version skew keeps old parents working), never 0. +func (b subagentWireBudget) fields() map[string]any { + out := make(map[string]any, 3) + if b.Seconds > 0 { + out["budget_seconds"] = b.Seconds + } + if b.Iterations > 0 { + out["budget_iterations"] = b.Iterations + } + if b.CostUSD > 0 { + out["budget_cost_usd"] = b.CostUSD + } + return out +} + +// subagentRiskCapOrder is the class set max_risk caps are expressed over, +// ordered by danger.Rank descending. It mirrors the class list +// clampClassesAboveMaxRisk walks (unread-script gating is enforced by the +// trust lockdown, not the max_risk cap, so UnreadExec is not part of cap +// semantics); extend both together if the class set grows. +var subagentRiskCapOrder = []danger.RiskClass{ + danger.Blocked, + danger.Destructive, + danger.Unknown, + danger.Persistence, + danger.SystemWrite, + danger.CodeExecution, + danger.NetworkEgress, + danger.Install, + danger.LocalWrite, + danger.Safe, +} + +// effectiveMaxRisk returns the child's effective post-clamp risk cap (P1): +// the highest-ranked class the resolved danger config does not outright +// deny, after the operator profile (applyProfile) and the trust lockdown +// (applySubagentTrust) ran. Under the default untrusted envelope this +// reports local_write — the operator's default envelope. Empty when every +// cap-expressible class is denied (total lockdown). +func effectiveMaxRisk(dc *danger.DangerousConfig) string { + if dc == nil { + return "" + } + for _, cls := range subagentRiskCapOrder { + if dc.ActionFor(cls) != danger.Deny { + return string(cls) + } + } + return "" +} + +// subagentCostEstimator prices cumulative token totals exactly the way +// /api/usage does (handleUsage): the per-million prices resolved for the +// run's model. configured() is handleUsage's prices_configured predicate — +// when no price side is configured the cost is unknowable and the wire +// omits it rather than emitting a fabricated $0 (odek never guesses +// provider prices). +type subagentCostEstimator struct { + inPerMillion float64 + outPerMillion float64 +} + +func newSubagentCostEstimator(limits budget.Limits, model string) subagentCostEstimator { + in, out := limits.ResolvePrices(model) + return subagentCostEstimator{inPerMillion: in, outPerMillion: out} +} + +func (e subagentCostEstimator) configured() bool { return e.inPerMillion > 0 || e.outPerMillion > 0 } + +func (e subagentCostEstimator) estimate(inputTokens, outputTokens int64) float64 { + return float64(inputTokens)/1e6*e.inPerMillion + float64(outputTokens)/1e6*e.outPerMillion +} + +// subagentWireContext carries everything the telemetry writer needs to +// decorate lifecycle records with the P1/P3/P6 wire fields: the resolved +// profile id ("" = built-in default envelope, omitted), the effective +// post-clamp risk cap, the effective budget block, the model-resolved cost +// estimator, and the engine usage probe for cost-so-far / final cost. +type subagentWireContext struct { + Profile string + MaxRisk string + Budget subagentWireBudget + Cost subagentCostEstimator + // Usage returns the engine's cumulative provider-reported (input, + // output) token totals — the same totals the budget Checker + // accumulates. Nil when no engine is attached yet. + Usage func() (int64, int64) } // ── Subagent Command ───────────────────────────────────────────────── @@ -370,6 +548,7 @@ type subagentResult struct { Denials []SubagentDenial `json:"denials,omitempty"` // policy denials observed (capped) DenialsTotal int `json:"denials_total,omitempty"` // total denials seen ParentSession string `json:"parent_session,omitempty"` // correlation id from --parent-session + CostUSD float64 `json:"cost_usd,omitempty"` // final server-side cost estimate (omitted when no prices configured) Artifacts []artifact.Ref `json:"artifacts,omitempty"` // odek.artifact-ref/v1 — runner-scanned, parent-validated } @@ -660,6 +839,27 @@ func subagentCmd(args []string) error { return fmt.Errorf("parent budget exhausted before start: %w", berr) } + // P1/P3/P6 wire context: everything the protocol-2 telemetry records + // and the result envelope report about this child's run posture. All + // values are RESOLVED — post operator-profile application (P4), post + // trust lockdown (P2/P3), and post task-budget clamp (M1.5). The Usage + // probe reads the engine's provider-reported cumulative totals so the + // cost fields use the exact /api/usage estimate; agent is assigned + // below, before any event can fire. + var agent *odek.Agent + wireCtx := subagentWireContext{ + Profile: profileName, + MaxRisk: effectiveMaxRisk(&resolved.Dangerous), + Budget: newSubagentWireBudget(resolved.Limits, cfg.timeout, cfg.maxIter), + Cost: newSubagentCostEstimator(resolved.Limits, resolved.Model), + Usage: func() (int64, int64) { + if agent == nil { + return 0, 0 + } + return int64(agent.TotalInputTokens()), int64(agent.TotalOutputTokens()) + }, + } + // The sub-agent system prompt is a FIXED constant — a trust boundary the // parent cannot write to. Parent-supplied goal/guidance/context are // delivered in the user request instead (fenced when untrusted), so they @@ -826,7 +1026,7 @@ func subagentCmd(args []string) error { } } } - agent, err := odek.New(aCfg) + agent, err = odek.New(aCfg) if err != nil { return fmt.Errorf("create agent: %w", err) } @@ -925,19 +1125,49 @@ func subagentCmd(args []string) error { } } + // P6: final estimated cost on the result envelope — the same + // /api/usage estimate (model-resolved per-million prices) over the + // engine's provider-reported token totals. Zero (omitted on the wire) + // when no price side is configured: clients must render cost as + // unavailable, never $0. + if agent != nil && wireCtx.Cost.configured() { + result.CostUSD = wireCtx.Cost.estimate(int64(agent.TotalInputTokens()), int64(agent.TotalOutputTokens())) + } + // Output JSON to stdout — the envelope is emitted exactly once, here. // Protocol-2 children emit a compact subagent_finished record followed // by a FRAMED result ({"type":"result",…}) so the parent's parser // cannot confuse protocol traffic with the result. Legacy children // keep the bare indented encoding. if telemetry != nil { - telemetry.emit(map[string]any{ + fin := map[string]any{ "type": "subagent_finished", "status": result.Status, "iterations": result.Iterations, "duration_s": result.DurationSeconds, "tokens_used": result.TokensUsed, - }) + } + // Wire v2 (P6): final cost estimate — omitted when no price side is + // configured, never a fabricated $0. + if result.CostUSD > 0 { + fin["cost_usd"] = result.CostUSD + } + // Wire v2 (P4): terminal artifact metadata in the spec shape + // {id, path, bytes} — bounded metadata only; artifact content never + // rides the telemetry wire. Mirrors the framed envelope's refs so a + // client that only watches state frames still sees the artifact list. + if len(result.Artifacts) > 0 { + arts := make([]map[string]any, 0, len(result.Artifacts)) + for _, a := range result.Artifacts { + item := map[string]any{"id": a.ID, "path": a.URI} + if a.SizeBytes != nil { + item["bytes"] = *a.SizeBytes + } + arts = append(arts, item) + } + fin["artifacts"] = arts + } + telemetry.emit(fin) } if protocol2 { raw, merr := json.Marshal(result) diff --git a/cmd/odek/subagent_artifacts_test.go b/cmd/odek/subagent_artifacts_test.go index 7041a92..880a46f 100644 --- a/cmd/odek/subagent_artifacts_test.go +++ b/cmd/odek/subagent_artifacts_test.go @@ -254,3 +254,66 @@ func TestStore_Cleanup_CascadesArtifacts(t *testing.T) { t.Errorf("Cleanup (indexed path) must cascade per removed session: %v", cascaded) } } + +// ── P4/P6 child half: the result envelope carries cost + artifacts ─── + +// The framed result envelope carries the final estimated cost (P6) and the +// runner-scanned artifact refs (P4 — the registry the artifact_read surface +// resolves against). Artifact entries travel in the odek.artifact-ref/v1 +// shape the parent validates fail-closed: id + size_bytes (the wire's +// "id"/"bytes") plus the file:// uri. +func TestSubagentResult_EnvelopeCarriesCostAndArtifacts(t *testing.T) { + dir := t.TempDir() + const content = "# Report\nfindings here" + writeArtifactFile(t, dir, "report.md", content) + refs, flags := scanArtifacts(dir, 1<<20) + if len(refs) != 1 || len(flags) != 0 { + t.Fatalf("scan = %d refs, flags %v; want 1/none", len(refs), flags) + } + + res := subagentResult{Status: "success", Summary: "done", CostUSD: 0.42, Artifacts: refs} + raw, err := json.Marshal(res) + if err != nil { + t.Fatal(err) + } + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if m["cost_usd"] != 0.42 { + t.Errorf("cost_usd = %v, want 0.42 on the envelope", m["cost_usd"]) + } + arts, ok := m["artifacts"].([]any) + if !ok || len(arts) != 1 { + t.Fatalf("artifacts = %v, want exactly 1 ref", m["artifacts"]) + } + a := arts[0].(map[string]any) + if a["id"] != "report" { + t.Errorf("artifact id = %v, want report", a["id"]) + } + if a["size_bytes"] != float64(len(content)) { + t.Errorf("artifact size_bytes = %v, want %d (runner-measured)", a["size_bytes"], len(content)) + } + if a["schema"] != artifact.SchemaArtifactRef { + t.Errorf("artifact schema = %v, want %s (parent validates this shape)", a["schema"], artifact.SchemaArtifactRef) + } + if u, _ := a["uri"].(string); !strings.HasPrefix(u, "file://") { + t.Errorf("artifact uri = %v, want file:// prefix", a["uri"]) + } +} + +// An envelope with no cost (prices unconfigured) and no artifacts omits +// both fields entirely — never a $0 cost or an empty artifacts array. +func TestSubagentResult_EnvelopeOmitsCostAndArtifactsWhenEmpty(t *testing.T) { + raw, err := json.Marshal(subagentResult{Status: "success", Summary: "done"}) + if err != nil { + t.Fatal(err) + } + s := string(raw) + if strings.Contains(s, "cost_usd") { + t.Errorf("cost_usd must be omitted when unset: %s", s) + } + if strings.Contains(s, "artifacts") { + t.Errorf("artifacts must be omitted when the task staged nothing: %s", s) + } +} diff --git a/cmd/odek/subagent_profiles_test.go b/cmd/odek/subagent_profiles_test.go index 247e92e..00abb78 100644 --- a/cmd/odek/subagent_profiles_test.go +++ b/cmd/odek/subagent_profiles_test.go @@ -445,6 +445,55 @@ func TestDefaultProfileEnvelope_TrustedChildClamped(t *testing.T) { } } +// ── P1: effective post-clamp risk cap on the wire ──────────────────── + +// The untrusted trust lockdown denies everything above local_write, so the +// effective cap the child reports must be local_write — exactly the +// operator's default envelope. +func TestEffectiveMaxRisk_UntrustedClampReportsLocalWrite(t *testing.T) { + var dc danger.DangerousConfig + applySubagentTrust(&dc, "untrusted", "") + if got := effectiveMaxRisk(&dc); got != "local_write" { + t.Errorf("effectiveMaxRisk = %q, want local_write (untrusted lockdown caps at local_write)", got) + } +} + +// A profile's max_risk is the effective cap the child reports: classes +// ranked above it are denied by the clamp, the cap class itself is not. +func TestEffectiveMaxRisk_ProfileCapReportsCap(t *testing.T) { + var dc danger.DangerousConfig + applyProfile(&dc, config.ProfileConfig{MaxRisk: "code_execution"}) + applySubagentTrust(&dc, "trusted", "") + if got := effectiveMaxRisk(&dc); got != "code_execution" { + t.Errorf("effectiveMaxRisk = %q, want code_execution (profile cap)", got) + } +} + +// A task-file max_risk cap is reported the same way as a profile cap. +func TestEffectiveMaxRisk_TaskMaxRiskReportsCap(t *testing.T) { + var dc danger.DangerousConfig + applySubagentTrust(&dc, "trusted", "system_write") + if got := effectiveMaxRisk(&dc); got != "system_write" { + t.Errorf("effectiveMaxRisk = %q, want system_write (task cap)", got) + } +} + +// Total lockdown (every cap-expressible class denied) reports "" — the +// wire omits the field rather than inventing a class. nil configs do too. +func TestEffectiveMaxRisk_TotalLockdownEmpty(t *testing.T) { + var dc danger.DangerousConfig + dc.Classes = map[danger.RiskClass]danger.Action{} + for _, cls := range subagentRiskCapOrder { + dc.Classes[cls] = danger.Deny + } + if got := effectiveMaxRisk(&dc); got != "" { + t.Errorf("effectiveMaxRisk = %q, want empty under total lockdown", got) + } + if got := effectiveMaxRisk(nil); got != "" { + t.Errorf("effectiveMaxRisk(nil) = %q, want empty", got) + } +} + // TestDelegateTasks_UnknownProfileFailsWithoutSpawn pins parent-side // fail-closed: an unknown profile must fail the task BEFORE a child is // spawned. The marker file proves whether the mock child ever ran. diff --git a/cmd/odek/subagent_registry.go b/cmd/odek/subagent_registry.go index e757456..bbbea9e 100644 --- a/cmd/odek/subagent_registry.go +++ b/cmd/odek/subagent_registry.go @@ -23,25 +23,44 @@ import ( const ( // maxSubagentRegistryEntries bounds the ring; oldest entries are evicted. maxSubagentRegistryEntries = 256 - // maxSubagentRegistryGoalChars truncates stored goals. - maxSubagentRegistryGoalChars = 200 + // maxSubagentRegistryGoalChars truncates stored goals. Raised from 200 + // to 2048 (wire v2): the goal rides every subagent_state frame so + // clients (e.g. bodek) can render usable task text, not an ellipsis. + maxSubagentRegistryGoalChars = 2048 ) +// subagentArtifact is the bounded metadata block the registry and the +// subagent_state terminal frame carry for one artifact produced by a +// sub-agent (wire v2). Metadata only — content is never inlined; the +// model-facing render path keeps the fail-closed artifact.Validate gate. +type subagentArtifact struct { + ID string `json:"id"` + Path string `json:"path,omitempty"` + Bytes int64 `json:"bytes,omitempty"` +} + // subagentEntry is one delegated task's lifecycle record. type subagentEntry struct { - TaskID string `json:"task_id"` - RunKey string `json:"run_key"` - Goal string `json:"goal,omitempty"` - Status string `json:"status,omitempty"` - Phase string `json:"phase"` // started | active | finished - PID int `json:"pid,omitempty"` - StartedAt time.Time `json:"started_at"` - FinishedAt time.Time `json:"finished_at,omitempty"` - Iterations int `json:"iterations,omitempty"` - Step int `json:"step,omitempty"` - LastTool string `json:"last_tool,omitempty"` - DurationSeconds float64 `json:"duration_seconds,omitempty"` - TokensUsed int `json:"tokens_used,omitempty"` + TaskID string `json:"task_id"` + RunKey string `json:"run_key"` + Goal string `json:"goal,omitempty"` + Status string `json:"status,omitempty"` + Phase string `json:"phase"` // queued | started | active | finished + PID int `json:"pid,omitempty"` + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at,omitempty"` + Iterations int `json:"iterations,omitempty"` + Step int `json:"step,omitempty"` + LastTool string `json:"last_tool,omitempty"` + DurationSeconds float64 `json:"duration_seconds,omitempty"` + TokensUsed int `json:"tokens_used,omitempty"` + Profile string `json:"profile,omitempty"` // requested on queued; effective (post-clamp) once the child reports + MaxRisk string `json:"max_risk,omitempty"` // requested on queued; effective (post-clamp) once the child reports + BudgetSeconds int `json:"budget_seconds,omitempty"` + BudgetIterations int `json:"budget_iterations,omitempty"` + CostUSD float64 `json:"cost_usd,omitempty"` // cumulative child-reported spend + BudgetCostUSD float64 `json:"budget_cost_usd,omitempty"` // present only when the child reports a cost cap + Artifacts []subagentArtifact `json:"artifacts,omitempty"` // terminal metadata from the framed result envelope } var subagentReg = struct { @@ -52,21 +71,23 @@ var subagentReg = struct { byID: map[string]*subagentEntry{}, } -// subagentRegistryRecord inserts a new entry (or replaces by task_id) and -// evicts the oldest when the ring overflows. +// subagentRegistryRecord inserts a new entry (or merges into the existing +// one by task_id) and evicts the oldest when the ring overflows. func subagentRegistryRecord(e *subagentEntry) { subagentReg.mu.Lock() defer subagentReg.mu.Unlock() - if prev, ok := subagentReg.byID[e.TaskID]; ok { - *e = *prev // re-record keeps accumulated state - } e.TaskID = strings.TrimSpace(e.TaskID) if e.StartedAt.IsZero() { e.StartedAt = time.Now() } if prev, ok := subagentReg.byID[e.TaskID]; ok { - // Replace in place, preserving ring position. - *prev = *e + // Merge: the new record's set fields win, zero/empty fields keep + // the accumulated state. The v1 restore-old-into-new behavior made + // any re-record a no-op — the wire-v2 queued → started transition + // relies on the started record overwriting the declared identity + // (profile/max_risk) with the effective values and carrying the + // budgets in. + mergeEntry(prev, e) return } cp := *e @@ -79,6 +100,68 @@ func subagentRegistryRecord(e *subagentEntry) { } } +// mergeEntry folds a fresh record into the accumulated entry: every set +// field on src wins; zero/empty fields keep dst's accumulated value. +func mergeEntry(dst, src *subagentEntry) { + if src.Goal != "" { + dst.Goal = src.Goal + } + if src.RunKey != "" { + dst.RunKey = src.RunKey + } + if src.Status != "" { + dst.Status = src.Status + } + if src.Phase != "" { + dst.Phase = src.Phase + } + if src.PID != 0 { + dst.PID = src.PID + } + if !src.StartedAt.IsZero() { + dst.StartedAt = src.StartedAt + } + if !src.FinishedAt.IsZero() { + dst.FinishedAt = src.FinishedAt + } + if src.Iterations != 0 { + dst.Iterations = src.Iterations + } + if src.Step != 0 { + dst.Step = src.Step + } + if src.LastTool != "" { + dst.LastTool = src.LastTool + } + if src.DurationSeconds != 0 { + dst.DurationSeconds = src.DurationSeconds + } + if src.TokensUsed != 0 { + dst.TokensUsed = src.TokensUsed + } + if src.Profile != "" { + dst.Profile = src.Profile + } + if src.MaxRisk != "" { + dst.MaxRisk = src.MaxRisk + } + if src.BudgetSeconds != 0 { + dst.BudgetSeconds = src.BudgetSeconds + } + if src.BudgetIterations != 0 { + dst.BudgetIterations = src.BudgetIterations + } + if src.CostUSD != 0 { + dst.CostUSD = src.CostUSD + } + if src.BudgetCostUSD != 0 { + dst.BudgetCostUSD = src.BudgetCostUSD + } + if len(src.Artifacts) > 0 { + dst.Artifacts = src.Artifacts + } +} + // subagentRegistryUpdate applies fn to the entry with the given task_id // (no-op when absent). New entries are auto-created so a progress line that // races its started record still lands. @@ -116,16 +199,23 @@ func newSubagentTelemetryRelay(send func(v any) error, runKey string) func(taskI logRelay(taskIdx, taskID, line) var rec struct { - Type string `json:"type"` - TaskID string `json:"task_id,omitempty"` - PID int `json:"pid,omitempty"` - Goal string `json:"goal,omitempty"` - Status string `json:"status,omitempty"` - Step int `json:"step,omitempty"` - Tool string `json:"tool,omitempty"` - Iterations int `json:"iterations,omitempty"` - DurationS float64 `json:"duration_s,omitempty"` - TokensUsed int `json:"tokens_used,omitempty"` + Type string `json:"type"` + TaskID string `json:"task_id,omitempty"` + PID int `json:"pid,omitempty"` + Goal string `json:"goal,omitempty"` + Status string `json:"status,omitempty"` + Step int `json:"step,omitempty"` + Tool string `json:"tool,omitempty"` + Iterations int `json:"iterations,omitempty"` + DurationS float64 `json:"duration_s,omitempty"` + TokensUsed int `json:"tokens_used,omitempty"` + Profile string `json:"profile,omitempty"` + MaxRisk string `json:"max_risk,omitempty"` + BudgetSeconds int `json:"budget_seconds,omitempty"` + BudgetIterations int `json:"budget_iterations,omitempty"` + CostUSD float64 `json:"cost_usd,omitempty"` + BudgetCostUSD float64 `json:"budget_cost_usd,omitempty"` + Artifacts []subagentArtifact `json:"artifacts,omitempty"` } if err := json.Unmarshal([]byte(line), &rec); err != nil { return @@ -138,25 +228,63 @@ func newSubagentTelemetryRelay(send func(v any) error, runKey string) func(taskI } switch rec.Type { - case "subagent_started": - goal := redact.RedactSecrets(rec.Goal) - if len(goal) > maxSubagentRegistryGoalChars { - goal = goal[:maxSubagentRegistryGoalChars] - } + case "subagent_queued": + // Parent-synthesized (delegate_tasks pre-spawn): the task was + // accepted but has not been spawned — the concurrency limiter + // may still be holding it. profile/max_risk carry the DECLARED + // values; the child's started record overwrites them with the + // effective post-clamp values. subagentRegistryRecord(&subagentEntry{ TaskID: taskID, RunKey: runKey, - Goal: goal, - Phase: "started", - Status: "running", - PID: rec.PID, + Goal: redactGoal(rec.Goal), + Profile: rec.Profile, + MaxRisk: rec.MaxRisk, + Phase: "queued", + Status: "queued", StartedAt: time.Now(), }) + case "subagent_started": + subagentRegistryRecord(&subagentEntry{ + TaskID: taskID, + RunKey: runKey, + Goal: redactGoal(rec.Goal), + Phase: "started", + Status: "running", + PID: rec.PID, + StartedAt: time.Now(), + Profile: rec.Profile, + MaxRisk: rec.MaxRisk, + BudgetSeconds: rec.BudgetSeconds, + BudgetIterations: rec.BudgetIterations, + CostUSD: rec.CostUSD, + BudgetCostUSD: rec.BudgetCostUSD, + }) case "subagent_progress": subagentRegistryUpdate(taskID, func(e *subagentEntry) { e.Phase = "active" e.Step = rec.Step e.LastTool = rec.Tool + // Budget/identity fields are re-asserted (and may first + // arrive) on progress records; last report wins. + if rec.Profile != "" { + e.Profile = rec.Profile + } + if rec.MaxRisk != "" { + e.MaxRisk = rec.MaxRisk + } + if rec.BudgetSeconds > 0 { + e.BudgetSeconds = rec.BudgetSeconds + } + if rec.BudgetIterations > 0 { + e.BudgetIterations = rec.BudgetIterations + } + if rec.CostUSD > 0 { + e.CostUSD = rec.CostUSD + } + if rec.BudgetCostUSD > 0 { + e.BudgetCostUSD = rec.BudgetCostUSD + } }) case "subagent_finished": subagentRegistryUpdate(taskID, func(e *subagentEntry) { @@ -165,6 +293,15 @@ func newSubagentTelemetryRelay(send func(v any) error, runKey string) func(taskI e.Iterations = rec.Iterations e.DurationSeconds = rec.DurationS e.TokensUsed = rec.TokensUsed + if rec.CostUSD > 0 { + e.CostUSD = rec.CostUSD + } + if rec.BudgetCostUSD > 0 { + e.BudgetCostUSD = rec.BudgetCostUSD + } + if len(rec.Artifacts) > 0 { + e.Artifacts = rec.Artifacts + } e.FinishedAt = time.Now() }) if rec.Status == "success" || rec.Status == "partial" { @@ -173,6 +310,26 @@ func newSubagentTelemetryRelay(send func(v any) error, runKey string) func(taskI subagentStats.failed.Add(1) } subagentStats.tokens.Add(int64(rec.TokensUsed)) + if rec.CostUSD > 0 { + subagentStats.costMicros.Add(int64(rec.CostUSD * 1e6)) + } + case "subagent_result": + // Parent-synthesized refinement: cost + artifact metadata + // extracted from the framed result envelope (the envelope + // itself never reaches this relay). Merge-only — phase, + // status, and lifetime counters stay owned by the + // subagent_finished / done-relay paths. + subagentRegistryUpdate(taskID, func(e *subagentEntry) { + if rec.CostUSD > 0 { + e.CostUSD = rec.CostUSD + } + if rec.BudgetCostUSD > 0 { + e.BudgetCostUSD = rec.BudgetCostUSD + } + if len(rec.Artifacts) > 0 { + e.Artifacts = rec.Artifacts + } + }) default: return // tool_call/tool_result/unknown — log-only } @@ -182,6 +339,16 @@ func newSubagentTelemetryRelay(send func(v any) error, runKey string) func(taskI } } +// redactGoal redacts and caps a goal for storage/relay (operator surface, +// but model-controlled text — same treatment as the log relay). +func redactGoal(goal string) string { + goal = redact.RedactSecrets(goal) + if len(goal) > maxSubagentRegistryGoalChars { + goal = goal[:maxSubagentRegistryGoalChars] + } + return goal +} + // subagentRegistryEmitState fans a task's current registry entry out to // the UI as a subagent_state WS message. Shared by the line relay // (child-driven transitions) and the done relay (parent-driven terminal @@ -190,7 +357,7 @@ func subagentRegistryEmitState(send func(v any) error, taskID string, taskIdx in snap := subagentRegistrySnapshot("") for _, e := range snap { if e.TaskID == taskID { - _ = send(map[string]any{ + msg := map[string]any{ "type": "subagent_state", "task_id": e.TaskID, "task_idx": taskIdx, @@ -202,7 +369,34 @@ func subagentRegistryEmitState(send func(v any) error, taskID string, taskIdx in "tool": e.LastTool, "duration_seconds": e.DurationSeconds, "tokens_used": e.TokensUsed, - }) + } + // Wire v2 fields — omitted when unset (0/empty), matching the + // registry entry's omitempty JSON contract. + if e.Goal != "" { + msg["goal"] = e.Goal + } + if e.Profile != "" { + msg["profile"] = e.Profile + } + if e.MaxRisk != "" { + msg["max_risk"] = e.MaxRisk + } + if e.BudgetSeconds > 0 { + msg["budget_seconds"] = e.BudgetSeconds + } + if e.BudgetIterations > 0 { + msg["budget_iterations"] = e.BudgetIterations + } + if e.CostUSD > 0 { + msg["cost_usd"] = e.CostUSD + } + if e.BudgetCostUSD > 0 { + msg["budget_cost_usd"] = e.BudgetCostUSD + } + if len(e.Artifacts) > 0 { + msg["artifacts"] = e.Artifacts + } + _ = send(msg) break } } @@ -296,14 +490,21 @@ var subagentStats struct { completed atomic.Int64 failed atomic.Int64 tokens atomic.Int64 + // costMicros accumulates the child-reported final cost estimates in + // micro-USD (int64 atomic — float atomics are not available). Exposed + // as the subagent_cost_usd sub-total so clients can render a total + // including sub-agents without client-side arithmetic. + costMicros atomic.Int64 } // subagentStatsSnapshot returns the lifetime sub-agent counters plus the -// number of currently non-finished registry entries. +// number of currently non-finished registry entries. Queued entries +// (accepted but not yet spawned) count as neither active nor finished. func subagentStatsSnapshot() map[string]any { active := 0 for _, e := range subagentRegistrySnapshot("") { - if e.Phase != "finished" { + // queued tasks are accepted-but-not-spawned: neither live nor done. + if e.Phase == "started" || e.Phase == "active" { active++ } } @@ -312,6 +513,7 @@ func subagentStatsSnapshot() map[string]any { "failed": subagentStats.failed.Load(), "active": active, "tokens_used": subagentStats.tokens.Load(), + "cost_usd": float64(subagentStats.costMicros.Load()) / 1e6, } } diff --git a/cmd/odek/subagent_telemetry_test.go b/cmd/odek/subagent_telemetry_test.go index 496e55a..f614915 100644 --- a/cmd/odek/subagent_telemetry_test.go +++ b/cmd/odek/subagent_telemetry_test.go @@ -1,10 +1,12 @@ package main import ( + "bytes" "encoding/json" "strings" "testing" + "github.com/BackendStack21/odek/internal/budget" "github.com/BackendStack21/odek/internal/events" ) @@ -237,3 +239,205 @@ func TestNewSubagentTelemetryWriter_NilWithoutTaskID(t *testing.T) { t.Error("writer must be nil without a task_id (standalone runs stay silent)") } } + +// ── Wire additions (P1/P3 — child half) ────────────────────────────── + +// wireHarness builds a writer with a fully configured wire context and a +// usage probe that counts consultations (the cost fields must be derived +// from the engine's cumulative totals, not guessed). +func wireHarness(t *testing.T, buf *bytes.Buffer) (*subagentTelemetryWriter, *int) { + t.Helper() + calls := 0 + tw := newSubagentTelemetryWriterWithWire(buf, "task-w1", subagentWireContext{ + Profile: "reviewer", + MaxRisk: "local_write", + Budget: newSubagentWireBudget( + budget.Limits{MaxCostUSD: 0.5, InputCostPerMillionUSD: 1.5, OutputCostPerMillionUSD: 7.5}, + 120, 15, + ), + Cost: subagentCostEstimator{inPerMillion: 1.5, outPerMillion: 7.5}, + Usage: func() (int64, int64) { + calls++ + return 2_000_000, 1_000_000 // 2M in, 1M out → 2*1.5 + 1*7.5 = 10.5 + }, + }) + if tw == nil { + t.Fatal("wire writer is nil") + } + return tw, &calls +} + +func decodeRecord(t *testing.T, buf *bytes.Buffer) map[string]any { + t.Helper() + line := buf.String() + var m map[string]any + if err := json.Unmarshal([]byte(strings.TrimSpace(line)), &m); err != nil { + t.Fatalf("record is not valid JSON: %v (%q)", err, line) + } + return m +} + +// P1: the started record carries the resolved profile id, the effective +// post-clamp risk cap, and the P3 budget block. cost_usd is deliberately +// absent — nothing has been spent at start. +func TestSubagentWire_StartedCarriesProfileRiskAndBudgets(t *testing.T) { + var buf bytes.Buffer + tw, _ := wireHarness(t, &buf) + tw.emitStarted(4242, 1, 120, 15) + + m := decodeRecord(t, &buf) + if m["task_id"] != "task-w1" { + t.Errorf("task_id = %v, want task-w1", m["task_id"]) + } + if m["profile"] != "reviewer" { + t.Errorf("profile = %v, want reviewer (resolved profile id)", m["profile"]) + } + if m["max_risk"] != "local_write" { + t.Errorf("max_risk = %v, want local_write (effective post-clamp cap)", m["max_risk"]) + } + if m["budget_seconds"] != float64(120) { + t.Errorf("budget_seconds = %v, want 120", m["budget_seconds"]) + } + if m["budget_iterations"] != float64(15) { + t.Errorf("budget_iterations = %v, want 15", m["budget_iterations"]) + } + if m["budget_cost_usd"] != 0.5 { + t.Errorf("budget_cost_usd = %v, want 0.5 (enforced cap)", m["budget_cost_usd"]) + } + if _, has := m["cost_usd"]; has { + t.Error("started record must not carry cost_usd (nothing spent at start)") + } +} + +// All-zero wire context (no profile, no cap, unconfigured budgets) must +// omit every new field — an absent field means "not configured", never 0. +func TestSubagentWire_StartedOmitsUnconfiguredWireFields(t *testing.T) { + var buf bytes.Buffer + tw := newSubagentTelemetryWriterWithWire(&buf, "task-w2", subagentWireContext{}) + if tw == nil { + t.Fatal("writer is nil") + } + tw.emitStarted(7, 0, 0, 0) + + m := decodeRecord(t, &buf) + for _, k := range []string{"profile", "max_risk", "budget_seconds", "budget_iterations", "budget_cost_usd"} { + if _, has := m[k]; has { + t.Errorf("%s must be omitted when unconfigured, got %v", k, m[k]) + } + } +} + +// P3: progress records carry the cumulative cost estimate (the same +// /api/usage math over the engine's provider-reported totals) plus the +// budget block, so a client can render % budget used per step. +func TestSubagentWire_ProgressCarriesCostSoFarAndBudgets(t *testing.T) { + var buf bytes.Buffer + tw, calls := wireHarness(t, &buf) + tw.emitProgress("read_file") + + m := decodeRecord(t, &buf) + if m["step"] != float64(1) || m["tool"] != "read_file" { + t.Errorf("progress core fields wrong: %v", m) + } + if m["cost_usd"] != 10.5 { + t.Errorf("cost_usd = %v, want 10.5 (2M in × $1.5/M + 1M out × $7.5/M)", m["cost_usd"]) + } + if m["budget_seconds"] != float64(120) || m["budget_iterations"] != float64(15) || m["budget_cost_usd"] != 0.5 { + t.Errorf("progress budget block wrong: %v", m) + } + if *calls != 1 { + t.Errorf("usage probe consulted %d times, want 1 (cost derived from engine totals)", *calls) + } +} + +// Without configured prices the cumulative cost is unknowable — the wire +// omits cost_usd entirely rather than emitting a fabricated $0. +func TestSubagentWire_ProgressOmitsCostWhenPricesAbsent(t *testing.T) { + var buf bytes.Buffer + tw := newSubagentTelemetryWriterWithWire(&buf, "task-w3", subagentWireContext{ + Budget: subagentWireBudget{Seconds: 60, Iterations: 10}, + Usage: func() (int64, int64) { return 999, 999 }, + }) + tw.emitProgress("shell") + + m := decodeRecord(t, &buf) + if _, has := m["cost_usd"]; has { + t.Errorf("cost_usd must be omitted without configured prices, got %v", m["cost_usd"]) + } + if m["budget_seconds"] != float64(60) || m["budget_iterations"] != float64(10) { + t.Errorf("budget block must survive without prices: %v", m) + } +} + +// The terminal record carries the FINAL cost estimate alongside the +// existing status/iterations/duration/tokens fields. +func TestSubagentWire_FinishedCarriesFinalCost(t *testing.T) { + var buf bytes.Buffer + tw := newSubagentTelemetryWriterWithWire(&buf, "task-w4", subagentWireContext{ + Cost: subagentCostEstimator{inPerMillion: 1.5, outPerMillion: 7.5}, + Usage: func() (int64, int64) { return 3_000_000, 1_000_000 }, // 4.5 + 7.5 = 12 + }) + tw.emitFinished("success", 4, 12.5, 900) + + m := decodeRecord(t, &buf) + if m["status"] != "success" || m["iterations"] != float64(4) || m["duration_s"] != 12.5 || m["tokens_used"] != float64(900) { + t.Errorf("finished core fields wrong: %v", m) + } + if m["cost_usd"] != 12.0 { + t.Errorf("cost_usd = %v, want 12 (3M in × $1.5/M + 1M out × $7.5/M)", m["cost_usd"]) + } +} + +func TestSubagentWire_FinishedOmitsCostWithoutPrices(t *testing.T) { + var buf bytes.Buffer + tw := newSubagentTelemetryWriterWithWire(&buf, "task-w5", subagentWireContext{}) + tw.emitFinished("error", 0, 1, 1) + + m := decodeRecord(t, &buf) + if _, has := m["cost_usd"]; has { + t.Error("cost_usd must be omitted without configured prices") + } +} + +// The estimator must reproduce the /api/usage math exactly: prices +// resolved for the run's model (per-model entry overriding the flat pair +// per field), spend = in/1e6·inPrice + out/1e6·outPrice, and configured() +// mirroring handleUsage's prices_configured predicate. +func TestSubagentCostEstimator_MatchesUsageAPIEstimate(t *testing.T) { + limits := budget.Limits{ + InputCostPerMillionUSD: 1, + OutputCostPerMillionUSD: 2, + ModelPrices: map[string]budget.ModelPrice{"m-fast": {InputCostPerMillionUSD: 5}}, + } + est := newSubagentCostEstimator(limits, "m-fast") + const in, out = 2_000_000, 3_000_000 + want := limits.ResolveForModel("m-fast").EstimatedCostUSD(in, out) + if got := est.estimate(in, out); got != want { + t.Errorf("estimate = %v, want %v (must equal the /api/usage math)", got, want) + } + if want != 16.0 { + t.Errorf("sanity: want 16 (2×5 + 3×2), got %v", want) + } + if !est.configured() { + t.Error("configured() = false with resolved prices, want true") + } + if inP, outP := limits.ResolvePrices("m-fast"); !(inP > 0 || outP > 0) != !est.configured() { + t.Error("configured() must mirror handleUsage's prices_configured predicate") + } + if (newSubagentCostEstimator(budget.Limits{}, "m-fast")).configured() { + t.Error("configured() = true with no prices, want false") + } +} + +// A cost cap without configured prices is NOT enforced (budget contract) — +// the wire must not report it as one. +func TestSubagentWireBudget_CostCapWithoutPricesNeverEmitted(t *testing.T) { + b := newSubagentWireBudget(budget.Limits{MaxCostUSD: 0.5}, 60, 10) + f := b.fields() + if _, has := f["budget_cost_usd"]; has { + t.Errorf("budget_cost_usd = %v, want omitted (CostEnforcementActive is false without prices)", f["budget_cost_usd"]) + } + if f["budget_seconds"] != 60 || f["budget_iterations"] != 10 { + t.Errorf("seconds/iterations lost: %v", f) + } +} diff --git a/cmd/odek/subagent_tool.go b/cmd/odek/subagent_tool.go index 9cba964..d9d3465 100644 --- a/cmd/odek/subagent_tool.go +++ b/cmd/odek/subagent_tool.go @@ -120,6 +120,30 @@ type delegateTasksTool struct { // tool's process can resolve it — same model as session prompt cancels. // Returns false when no such task is live (unknown id, already finished, // or the child exited before the stop arrived). +// emitSubagentQueued records + fans out the queued phase for a task that +// delegate_tasks accepted but has not spawned yet (concurrency ceiling). +// It synthesizes the same record shape the telemetry line relay parses, so +// registry state and WS subagent_state fan-out stay on one path. profile +// and max_risk carry the DECLARED values here; the child's started record +// overwrites them with the effective post-clamp values. +func (t *delegateTasksTool) emitSubagentQueued(taskIdx int, taskID, goal, profile, maxRisk string) { + if t.OnSubagentLog == nil { + return // no wire attached (bare-struct tests, non-serve runs) + } + rec := map[string]any{"type": "subagent_queued", "goal": goal} + if profile != "" { + rec["profile"] = profile + } + if maxRisk != "" { + rec["max_risk"] = maxRisk + } + b, err := json.Marshal(rec) + if err != nil { + return + } + t.OnSubagentLog(taskIdx, taskID, string(b)) +} + func (t *delegateTasksTool) CancelTask(taskID string) bool { return cancelSubagentTask(taskID) } @@ -264,11 +288,14 @@ func (t *delegateTasksTool) Call(args string) (string, error) { t.eventMu.Unlock() for i, task := range input.Tasks { - t.acquireSem(sem, emitFn, i) // Task id is minted HERE (not inside runTask) so the per-task // artifact dir can be created serially and correlated with the id // the child echoes on every telemetry record. taskID := newTaskID() + // Wire v2 (P2): record + emit the queued phase BEFORE acquiring a + // limiter slot, so clients see every accepted task immediately — + // including the ones still waiting for a concurrency slot. + t.emitSubagentQueued(i, taskID, task.Goal, task.Profile, task.MaxRisk) // Per-task artifact dir is created serially (MkdirAll idempotent, but // serial keeps 0700 semantics obvious) and captured by the goroutine. // Skipped entirely when no artifacts root is wired (bare-struct tests). @@ -277,6 +304,7 @@ func (t *delegateTasksTool) Call(args string) (string, error) { dirs[i] = d } } + t.acquireSem(sem, emitFn, i) run := t.runTaskFn if run == nil { run = t.runTask diff --git a/cmd/odek/subagent_wire_v2_test.go b/cmd/odek/subagent_wire_v2_test.go new file mode 100644 index 0000000..caf0060 --- /dev/null +++ b/cmd/odek/subagent_wire_v2_test.go @@ -0,0 +1,157 @@ +package main + +import ( + "encoding/json" + "sync" + "testing" +) + +// Wire v2 (P2/P6) — queued phase + cost sub-total. These pin the +// delegate_tasks pre-spawn emission path and the /api/usage sub-agent +// cost sub-total introduced with the sub-agent wire contract. + +// TestSubagentRegistry_QueuedPhaseFlow pins the queued → started → active +// phase progression through the telemetry relay: a parent-synthesized +// subagent_queued record records the DECLARED profile/max_risk, the child's +// started record overwrites them with the effective post-clamp values, and +// every transition fans out a subagent_state frame carrying the wire-v2 +// identity fields. +func TestSubagentRegistry_QueuedPhaseFlow(t *testing.T) { + var mu sync.Mutex + var frames []map[string]any + send := func(v any) error { + mu.Lock() + defer mu.Unlock() + if m, ok := v.(map[string]any); ok && m["type"] == "subagent_state" { + // The relay fans log lines through the same send path; only the + // state transitions are under test here. + frames = append(frames, m) + } + return nil + } + relay := newSubagentTelemetryRelay(send, "run-q1") + + feed := func(line string) { + t.Helper() + relay(0, "task-q1", line) + } + + // Parent-side queued record (synthesized by emitSubagentQueued). + feed(`{"type":"subagent_queued","task_id":"task-q1","goal":"audit the auth flow","profile":"reviewer","max_risk":"read_only"}`) + // Child-side started record with EFFECTIVE post-clamp identity. + feed(`{"type":"subagent_started","task_id":"task-q1","goal":"audit the auth flow","pid":4242,"profile":"default","max_risk":"local_write","budget_seconds":120,"budget_iterations":15,"budget_cost_usd":0.5}`) + // Child-side progress with cumulative cost. + feed(`{"type":"subagent_progress","task_id":"task-q1","step":3,"tool":"read_file","iterations":3,"cost_usd":0.02}`) + + mu.Lock() + defer mu.Unlock() + if len(frames) != 3 { + t.Fatalf("frames = %d, want 3 (queued/started/progress)", len(frames)) + } + q, s, a := frames[0], frames[1], frames[2] + + if q["phase"] != "queued" { + t.Errorf("queued frame phase = %v, want queued", q["phase"]) + } + if q["goal"] != "audit the auth flow" || q["profile"] != "reviewer" || q["max_risk"] != "read_only" { + t.Errorf("queued identity = %v/%v/%v, want declared values", q["goal"], q["profile"], q["max_risk"]) + } + if s["phase"] != "started" { + t.Errorf("started frame phase = %v, want started", s["phase"]) + } + // Effective (post-clamp) identity must overwrite the declared values. + if s["profile"] != "default" || s["max_risk"] != "local_write" { + t.Errorf("started identity = %v/%v, want effective default/local_write", s["profile"], s["max_risk"]) + } + if s["budget_seconds"] != 120 || s["budget_iterations"] != 15 { + t.Errorf("started budgets = %v/%v, want 120/15", s["budget_seconds"], s["budget_iterations"]) + } + if s["budget_cost_usd"] != 0.5 { + t.Errorf("started budget_cost_usd = %v, want 0.5", s["budget_cost_usd"]) + } + if a["phase"] != "active" { + t.Errorf("progress frame phase = %v, want active", a["phase"]) + } + if a["cost_usd"] != 0.02 { + t.Errorf("progress cost_usd = %v, want 0.02 (cumulative)", a["cost_usd"]) + } + + // Registry snapshot carries the same truth (reload half). + for _, e := range subagentRegistrySnapshot("run-q1") { + if e.TaskID != "task-q1" { + continue + } + if e.Phase != "active" || e.Profile != "default" || e.MaxRisk != "local_write" || + e.BudgetSeconds != 120 || e.BudgetIterations != 15 || e.CostUSD != 0.02 { + t.Errorf("snapshot entry = %+v", e) + } + return + } + t.Fatal("task-q1 missing from registry snapshot") +} + +func TestSubagentRegistry_QueuedFinishedCarriesCostAndArtifacts(t *testing.T) { + var got []map[string]any + relay := newSubagentTelemetryRelay(func(v any) error { + if m, ok := v.(map[string]any); ok { + got = append(got, m) + } + return nil + }, "run-q2") + + relay(1, "task-q2", `{"type":"subagent_started","task_id":"task-q2","goal":"g","pid":7}`) + relay(1, "task-q2", `{"type":"subagent_finished","task_id":"task-q2","status":"success","iterations":9,"duration_s":12.5,"tokens_used":4200,"cost_usd":0.42,"artifacts":[{"id":"report","path":"file:///tmp/report.md","bytes":2048}]}`) + + var fin map[string]any + for _, m := range got { + if m["phase"] == "finished" { + fin = m + } + } + if fin == nil { + t.Fatal("no finished frame emitted") + } + if fin["cost_usd"] != 0.42 { + t.Errorf("finished cost_usd = %v, want 0.42", fin["cost_usd"]) + } + arts, ok := fin["artifacts"].([]subagentArtifact) + if !ok || len(arts) != 1 { + t.Fatalf("finished artifacts = %v (%T), want one subagentArtifact entry", fin["artifacts"], fin["artifacts"]) + } + if arts[0].ID != "report" || arts[0].Bytes != 2048 { + t.Errorf("artifact[0] = %+v, want id=report bytes=2048", arts[0]) + } + + for _, e := range subagentRegistrySnapshot("run-q2") { + if e.TaskID == "task-q2" && (e.CostUSD != 0.42 || len(e.Artifacts) != 1) { + t.Errorf("snapshot entry cost/artifacts = %v/%v, want 0.42/1", e.CostUSD, len(e.Artifacts)) + } + } +} + +func TestDelegateTasks_EmitSubagentQueued(t *testing.T) { + var lines []string + tt := &delegateTasksTool{ + OnSubagentLog: func(taskIdx int, taskID, line string) { + lines = append(lines, line) + }, + } + tt.emitSubagentQueued(2, "task-q3", "audit the auth flow", "reviewer", "read_only") + if len(lines) != 1 { + t.Fatalf("emitted %d records, want 1", len(lines)) + } + var m map[string]any + if err := json.Unmarshal([]byte(lines[0]), &m); err != nil { + t.Fatalf("record not JSON: %v", err) + } + if m["type"] != "subagent_queued" || m["goal"] != "audit the auth flow" || + m["profile"] != "reviewer" || m["max_risk"] != "read_only" { + t.Errorf("queued record = %v", m) + } + // Correlation rides the positional taskID argument (the relay binds it), + // so the record itself only needs the identity fields. + + // Nil relay (bare-struct tests, non-serve runs) must be a no-op. + bare := &delegateTasksTool{} + bare.emitSubagentQueued(0, "task-q4", "g", "", "") +} diff --git a/docs/EXTENSIONS.md b/docs/EXTENSIONS.md index d90bc55..7dd2934 100644 --- a/docs/EXTENSIONS.md +++ b/docs/EXTENSIONS.md @@ -279,7 +279,14 @@ price in the entry falls back individually); unknown models use the flat pair. Budget enforcement is wired into `odek run` and `odek subagent` — the sub-agent enforces the operator limits (clamped by any parent-inherited budget when `subagent.budget_inherit` is `"share"`) and reports -`status: "budget_exhausted"` with exit code 4. `continue`, REPL, `serve`, and +`status: "budget_exhausted"` with exit code 4. In share mode the task +file's `budget` block carries the parent's remaining `max_runtime_seconds`, +`max_tool_calls`, and `max_cost_usd` plus explicit `runtime_exhausted`, +`tool_calls_exhausted`, and `cost_exhausted` flags — a remaining of 0 is +wire-ambiguous with an unconfigured limit, the flags are not. An exhausted +parent dimension is a hard cap of 0 for the child: the spawn fails fast +with the typed budget error and exit code 4 instead of starting an +unbounded child. `continue`, REPL, `serve`, and Telegram do not yet enforce limits, and there is no `ODEK_*` env-var layer for limits — sources are the `limits` config section (project configs may only *lower* global values, and project-set prices — flat or per-model — are diff --git a/docs/SUBAGENTS.md b/docs/SUBAGENTS.md index b6360c7..f5382f8 100644 --- a/docs/SUBAGENTS.md +++ b/docs/SUBAGENTS.md @@ -493,6 +493,60 @@ artifact_read({ "id": "report", "offset": 65536 }) # continue paging Lifecycle: deleting a session deletes its artifacts (all paths — CLI, API, Telegram, retention sweep); the storage janitor backstop sweeps orphans after `maintenance.artifacts_max_age_hours` (default 24 hours, `0` = keep forever). +## Wire contract (v2) + +Clients (Web UI, bodek) consume sub-agent telemetry through three surfaces: +live `subagent_state` WS frames, `GET /api/subagents` (registry snapshot — +same fields as the frames), and the framed result envelope. All v2 fields are +`omitempty`: old clients ignore them, and absent means "unavailable", never +zero. + +**Phases.** A task's `phase` moves `queued → started → active → finished`. +`queued` is emitted by the parent the moment `delegate_tasks` accepts a task — +before it acquires a concurrency slot — so an 8-task delegation on a 2-slot +operator reads `2 live · 6 queued`. Queued tasks count as neither active nor +finished in the stats. + +**State frame fields.** In addition to the v1 fields (`task_id`, `task_idx`, +`run_key`, `phase`, `status`, `step`, `iterations`, `tool`, +`duration_seconds`, `tokens_used`): + +| Field | Carries | Notes | +|---|---|---| +| `goal` | task goal text | redacted, clamped to 2048 chars server-side (model-controlled input) | +| `profile` | profile id | declared value while queued; the child's effective (post-clamp) value from `started` on | +| `max_risk` | effective risk ceiling | declared while queued; effective (post operator-profile resolution) from `started` | +| `budget_seconds`, `budget_iterations` | enforced caps | 0/absent = uncapped; present on `started` and `active` | +| `budget_cost_usd` | enforced cost cap | only when cost enforcement is active (cap + resolved prices) | +| `cost_usd` | server-side cost estimate | cumulative on live frames; final on `finished` | +| `artifacts` | `[{id, path, bytes}]` | terminal only — metadata from the result envelope; content never rides the wire | + +The framed result envelope adds `cost_usd` (final) and `artifacts` +(the full `odek.artifact-ref/v1` refs — a superset of the frame metadata). + +**Cost semantics (authoritative — do not re-derive).** `cost_usd` values are +computed server-side with the exact `/api/usage` math (per-million prices +resolved for the child's model over provider-reported token totals). Zero or +absent means "prices not configured" — clients must render cost as +unavailable, never `$0`. Totals are split, not folded: the serve-lifetime +`tokens_in`/`tokens_out`/`estimated_cost_usd` in `/api/usage` cover **parent +turns only** (children are separate odek processes with their own sessions); +sub-agent spend is broken out under `subagents.tokens_used` and, as of v2, +`subagents.cost_usd` (lifetime sum of final per-task estimates). A client +total including sub-agents is therefore `estimated_cost_usd + +subagents.cost_usd` — or, per batch, the sum of `cost_usd` over final +result envelopes only (cumulative state-frame values double-count on +replay). + +**Blocking model — deny, never prompt.** Sub-agents can never park waiting +on an approval: the child runs non-interactive with `deny` forced for every +operation that would prompt, and denied operations are listed in the result's +`denials` array (capped, with a separate total). There is no +`waiting_approval` state and none is planned; a card showing `running` with +denied operations is working through them or will finish with them reported. +Clients should treat `error` / `timeout` / `cancelled` as the only sticky +outcomes. + ## Tips - **Keep goals small** — one file, one concern per sub-agent. If a goal spans 3 files, it's probably not a good decomposition boundary.