From 358dd3d262685d528e070aadfbc261fb7879abf1 Mon Sep 17 00:00:00 2001 From: ailang-agent Date: Thu, 13 Aug 2026 13:13:02 +0000 Subject: [PATCH 1/2] feat(observatory): mission stage accounting + node-generic cloud routing (M2+M3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes M-MISSION-LOOP-UNIFIED-TELEMETRY. M1 (session-keyed chain linkage) landed in 56b449d01; this is M2 and M3. M2 — a mission iteration reported $0.0000 while holding $0.1077. Two defects with DIFFERENT owners, which is why they are separate changes: * writer-side: PostIteration created stages and never transitioned them, so every stage kept CreateStage's pending default. IterationStage now carries a per-stage `status`. * caller-side: IterationStage already carried TokensIn/TokensOut and already passed them to UpdateStageMetrics — the zeros came from the poster. The mission-control skill now supplies the real counts. Stage cost/tokens are also aggregated into the chain total, which nothing credited before. `status` is per stage on purpose: a stage that failed reads back `failed`, and StagesCompleted excludes it. Blanket-completing an iteration's stages would satisfy "nothing left pending" while hiding the failure, so that is tested against rather than merely avoided. An omitted status stays valid and leaves the stage pending (the skill and the CLI ship independently); an unrecognised value is rejected loudly rather than coerced. M3 — PostIteration now writes through a narrow IterationSink, a strict subset of observatory.Backend, so the Firestore store satisfies it (compile-time assertion added there). UpdateStageEvalAssessment is SQLite-only, so it is an optional IterationModelSink upgrade whose absence is reported, never silent. PostToLegs/FlushLegs write to every destination this NODE has configured: local always, cloud when AILANG_STORAGE=gcp. Nothing is rig-specific. hybrid is deliberately not a cloud leg — it resolves the observatory to local SQLite, so treating it as one would dual-write the same database twice. Each leg spools SEPARATELY. A shared spool would replay a post the local leg already stored, duplicating that chain on every flush. This extends the existing bounded+loud spool rather than inventing a fail-soft policy: the 100-entry / 1 MiB caps hold, every buffering event warns, and the command still exits 0. `ailang chains list|view --remote` reads the configured cloud observatory. Local stays the default; --remote with no cloud configured is an error, because a remote query silently answered from SQLite would report a cloud-side record as absent when it was never queried. Not done: the live cloud read-back, which needs credentials the sprint sandbox does not have. Recorded in the design doc and the handover. Tests: the SQLite read-back tests run in CI; every accounting rule is ALSO covered by cgo-free fake-sink tests, because this sandbox has no C toolchain. Co-Authored-By: Claude Opus 5 --- ...rint_M-MISSION-LOOP-UNIFIED-TELEMETRY.json | 95 +++++ .claude/skills/mission-control/SKILL.md | 39 ++- changelogs/v0.18-current.md | 40 +++ cmd/ailang/chains.go | 24 +- cmd/ailang/chains_post.go | 122 +++++-- cmd/ailang/chains_remote.go | 48 +++ cmd/ailang/chains_remote_test.go | 110 ++++++ cmd/ailang/chains_util.go | 9 +- cmd/ailang/help.go | 4 +- ...HANDOVER-mission-loop-unified-telemetry.md | 9 +- ...sion-loop-unified-telemetry-sprint-plan.md | 20 +- .../m-mission-loop-unified-telemetry.md | 37 +- internal/observatory/iteration_legs.go | 118 +++++++ internal/observatory/iteration_legs_test.go | 239 +++++++++++++ internal/observatory/iteration_post.go | 123 ++++++- internal/observatory/iteration_post_test.go | 234 +++++++++++++ internal/observatory/iteration_sink_test.go | 324 ++++++++++++++++++ internal/storage/firestore/observatory.go | 8 + 18 files changed, 1505 insertions(+), 98 deletions(-) create mode 100644 .ailang/state/sprints/sprint_M-MISSION-LOOP-UNIFIED-TELEMETRY.json create mode 100644 cmd/ailang/chains_remote.go create mode 100644 cmd/ailang/chains_remote_test.go create mode 100644 internal/observatory/iteration_legs.go create mode 100644 internal/observatory/iteration_legs_test.go create mode 100644 internal/observatory/iteration_post_test.go create mode 100644 internal/observatory/iteration_sink_test.go diff --git a/.ailang/state/sprints/sprint_M-MISSION-LOOP-UNIFIED-TELEMETRY.json b/.ailang/state/sprints/sprint_M-MISSION-LOOP-UNIFIED-TELEMETRY.json new file mode 100644 index 000000000..fe7676477 --- /dev/null +++ b/.ailang/state/sprints/sprint_M-MISSION-LOOP-UNIFIED-TELEMETRY.json @@ -0,0 +1,95 @@ +{ + "sprint_id": "M-MISSION-LOOP-UNIFIED-TELEMETRY", + "created": "2026-08-13", + "design_doc": "design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry.md", + "plan_path": "design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry-sprint-plan.md", + "handover": "design_docs/planned/v0_33_2/HANDOVER-mission-loop-unified-telemetry.md", + "branch": "coordinator/task-d98bb271", + "base_commit": "3941c0db", + "verify_profile": "go-compiler", + "github_issues": [], + "velocity": { + "target_loc_per_day": 250, + "estimated_total_loc": 500, + "estimated_days": 2 + }, + "sequencing": "M1 -> M2 -> M3 (M1 landed 56b449d01; M2 independently useful; M3 depends on M2's writer)", + "design_freeze": { + "dual_write_node_generic": "RATIFIED (Mark, 2026-08-13). Dual-write, not mirror: volume is 42 chains / 238 stages per day, so mirroring buys little and costs a reconciliation problem. Scope is NODE-GENERIC - 'this server, laptop, cloud, other nodes in the future'. Nothing may hardcode 'the rig'; the node is a parameter.", + "never_block_offline": "RATIFIED (Mark, 2026-08-13): 'no block if not available, at least until we harden availability.' Implement by EXTENDING the existing bounded+loud spool (internal/observatory/spool.go), not by inventing a policy. Availability hardening is deferred, not forgotten - the spool's bounded+loud contract is what keeps 'no block' honest rather than silently lossy.", + "local_analysis_reads_cloud_opt_in": "RATIFIED (Mark, 2026-08-13): yes, OPT-IN. OpenDefaultStore() keeps its local default so offline nodes stay first-class; an explicit remote mode is added. Opt-in is reversible - if --remote turns out to be always-passed, flipping the default later is one line with evidence behind it. Canonical-first is the version you cannot walk back." + }, + "reality_checks": { + "rc1_m2_two_defects_different_owners": "CONFIRMED - (a) writer-side: PostIteration (iteration_post.go:103-128) calls CreateStage -> UpdateStageMetrics -> UpdateStageEvalAssessment and NEVER calls UpdateStageStatus, so every stage keeps CreateStage's StageStatusPending default (store_chains.go:292). (b) caller-side: IterationStage ALREADY carries TokensIn/TokensOut/CostUSD and iteration_post.go:116 already passes them to UpdateStageMetrics - the zeros come from the poster (the mission-control skill).", + "rc2_chain_total_never_aggregated": "CONFIRMED - Store.UpdateChainMetrics exists (store_chains.go:227, increments total_cost/total_tokens/total_turns) and PostIteration never calls it. That is why iter-190 reads $0.0000 while holding $0.1077 of stage cost.", + "rc3_m3_single_hardcoded_backend": "CONFIRMED - cmd/ailang/chains_post.go:59 hardcodes NewSQLiteBackendFromPath(DefaultDatabasePath()) and is the SINGLE place the mission loop's backend is chosen. chainsPostIterationCommand ALREADY wraps that write in the bounded+loud spool, so the ratified never-block requirement is already satisfied structurally - M3 extends the wrapper, it does not build fail-soft from scratch.", + "rc4_eval_assessment_not_on_backend_iface": "CONFIRMED - UpdateStageEvalAssessment is a *Store method (store_chains_eval.go:14) and is NOT on the observatory.Backend interface, so the Firestore ObservatoryStore does not implement it. The dual-write sink must treat it as OPTIONAL (interface upgrade), not required - otherwise the cloud leg cannot compile against the same code path.", + "rc5_no_import_cycle_for_storage_selection": "CONFIRMED - internal/storage imports internal/observatory, so observatory CANNOT import storage. The cloud backend selection therefore belongs in cmd/ailang/chains_post.go (which is also where the single hardcoded backend lives), reusing internal/storage.NewBackends rather than adding a second selection mechanism." + }, + "features": [ + { + "id": "M1_SESSION_KEYED_CHAIN_LINKAGE", + "description": "Make a Broadcast span resolve to its chain: register the correlation session_id as a sessions row bound to chain_id + stage_id, and resolve chain_id via the sessions table in convertSpan when session.id is present and no explicit ailang.chain_id was supplied. ailang.chain_id keeps precedence.", + "estimated_loc": 190, + "dependencies": [], + "acceptance_criteria": [ + "An OTLP/JSON span carrying session.id matching a seeded session resolves to that session's chain_id and stage_id", + "NEGATIVE CONTROL: a Claude Code span carrying session.id and NO chain resolves exactly as it does today", + "An explicit ailang.chain_id WINS over a conflicting session.id lookup", + "A session.id with no matching row leaves chain_id empty and does not error", + "go test ./internal/observatory/... green" + ], + "passes": true, + "started": "2026-08-13T09:00:00Z", + "completed": "2026-08-13T12:00:00Z", + "notes": "Landed 56b449d01 (before this branch). LookupChainBySessionID added to the Backend interface + all impls; the chain-extraction block was patched in BOTH convertLogToSpan and convertSpan - Broadcast exports traces, not logs, so the logs-only first patch left the end-to-end test red. Covered by internal/observatory/session_chain_linkage_test.go." + }, + { + "id": "M2_MISSION_STAGE_ACCOUNTING", + "description": "Two defects with DIFFERENT owners. Writer-side: add a Status field to IterationStage and call UpdateStageStatus so stages stop reading pending. Caller-side: supply real token counts from the mission-control skill (the path is already wired). Then aggregate stage cost/tokens into the chain total via UpdateChainMetrics.", + "estimated_loc": 190, + "dependencies": [], + "acceptance_criteria": [ + "A posted stage with a terminal status reads back as that status, NOT pending", + "A failed stage reads back failed - blanket-completing every stage would hide real failures and is explicitly wrong", + "A stage posted with tokens reads back with those tokens", + "Chain total equals the sum of its stages - regression fixture built from the real iter-190 shape: 4 stages, 3 providers, two stages with cost-but-zero-tokens", + "A post that omits Status still works and defaults to today's behaviour (version skew between the skill and the CLI)" + ], + "passes": true, + "started": "2026-08-13T13:00:00Z", + "completed": "2026-08-13T13:40:00Z", + "notes": "Writer-side: IterationStage gained a per-stage Status (pending|running|awaiting_approval|completed|failed); PostIteration now calls UpdateStageStatus. Empty stays valid and leaves the stage pending (version skew with the mission-control skill); an unrecognised value is REJECTED by Validate rather than coerced. Aggregation: stage cost/tokens summed into UpdateChainMetrics, which nothing credited before (iter-190 read $0.0000 holding $0.1077). Caller-side: .claude/skills/mission-control/SKILL.md Gate 4 now supplies real tokens_in/tokens_out and a per-stage status, with the do-not-blanket-complete rule stated inline. Tests: iteration_post_test.go (SQLite read-back; runs in CI where cgo is on) + iteration_sink_test.go (same semantics through a fake sink, runs without cgo) incl. the iter-190 fixture, the failed-stage criterion, StagesCompleted excluding the failed stage, and call-ordering." + }, + { + "id": "M3_NODE_GENERIC_CLOUD_ROUTING", + "description": "Make the backend at chains_post.go:59 selectable rather than hardcoded SQLite, reusing internal/storage.NewBackends (which already resolves local/gcp/hybrid from AILANG_STORAGE). Dual-write local AND cloud per the ratified decision; extend the EXISTING spool to cover a cloud write failure; opt-in remote read for analysis with local staying the default.", + "estimated_loc": 120, + "dependencies": [ + "M2_MISSION_STAGE_ACCOUNTING" + ], + "acceptance_criteria": [ + "With cloud configured, a posted iteration appears in BOTH stores", + "Cloud unreachable -> the post is spooled, a loud stderr notice fires, and the command exits 0", + "The spool stays bounded - its existing 100-entry / 1 MiB caps still hold with cloud failures added", + "With no cloud configured, behaviour is byte-identical to today", + "Node-generic: nothing hardcodes the rig. The node is a parameter." + ], + "passes": true, + "started": "2026-08-13T13:40:00Z", + "completed": "2026-08-13T14:30:00Z", + "notes": "PostIteration refactored onto a narrow IterationSink (a strict subset of observatory.Backend, so the Firestore ObservatoryStore satisfies it \u2014 compile-time assertion added there). UpdateStageEvalAssessment is SQLite-only, so it is an OPTIONAL IterationModelSink upgrade whose absence is reported on stderr, never silent. New internal/observatory/iteration_legs.go: PostToLegs/FlushLegs write to every configured destination with PER-LEG spools \u2014 a shared spool would replay a post the local leg already stored and duplicate the chain on every flush. cmd/ailang/chains_post.go builds the legs: local always, cloud only when AILANG_STORAGE=gcp (hybrid resolves the observatory to local SQLite, so it is deliberately NOT a cloud leg). Nothing is rig-specific. Opt-in remote READ: --remote on chains list/view via openChainBackend; with no cloud configured it ERRORS rather than silently answering from SQLite. NOT DONE: the live cloud end-to-end confirmation, which needs credentials this sandbox does not have \u2014 recorded in the design doc and the handover." + } + ], + "definition_of_done": [ + "make build && make test green", + "go test ./internal/observatory/... ./cmd/ailang/... green", + "make lint && make fmt-check && make check-file-sizes && make check-boundaries green", + "A failed stage still reads back failed (no blanket-completion)", + "An unversioned payload (no status field) keeps working", + "CHANGELOG updated; design doc + sprint plan moved planned/ -> implemented/ on green" + ], + "risk_level": "medium", + "status": "completed", + "environment_note": "Executed in a sandbox with CGO_ENABLED=0 and no C toolchain, so every go-sqlite3-backed test in internal/observatory fails to open a database (177 such failures at baseline, ALL environmental \u2014 zero non-cgo failures before or after this sprint). The new SQLite read-back tests therefore compile and vet here but are first EXECUTED in CI. That is why each M2/M3 accounting rule is ALSO covered by a cgo-free fake-sink test that was run and passed locally." +} \ No newline at end of file diff --git a/.claude/skills/mission-control/SKILL.md b/.claude/skills/mission-control/SKILL.md index 89a1f0733..6ca3f807a 100644 --- a/.claude/skills/mission-control/SKILL.md +++ b/.claude/skills/mission-control/SKILL.md @@ -1748,24 +1748,49 @@ to the bounded, LOUD, fail-soft Go subcommand — NEVER inline shell spooling: ```bash # stages: metered lanes carry $ + model + tokens; quota lanes carry quota_bucket # (fable|opus|sonnet) and ZERO tokens/cost (subscription burn is bucket-visible, not dollar-faked). +# EVERY stage carries its OWN `status` — see the two rules below. cat <","cost_usd":,"tokens_in":,"tokens_out":}, - {"role":"controller","quota_bucket":"opus"}, - {"role":"evaluator","quota_bucket":"sonnet"} + {"role":"executor","provider":"codex","model":"","cost_usd":,"tokens_in":,"tokens_out":,"status":"completed"}, + {"role":"controller","quota_bucket":"opus","status":"completed"}, + {"role":"evaluator","quota_bucket":"sonnet","status":"failed"} ] } JSON ``` +**Two rules on this payload (M-MISSION-LOOP-UNIFIED-TELEMETRY M2), both measured defects:** + +1. **`status` is per stage, and a stage that failed says `failed`.** Vocabulary: `pending`, + `running`, `awaiting_approval`, `completed`, `failed`. Marking every stage `completed` because + the iteration ended would satisfy "no stage left pending" and *hide the failure* — that is the + one thing this field must not do. Omitting `status` is still accepted (older payloads keep + working) and leaves the stage `pending`; an unrecognised value is REJECTED loudly rather than + coerced. +2. **`tokens_in`/`tokens_out` must be the REAL counts for every metered lane.** Measured on + `manual:mission:v1/iter-190`: the two quorum stages posted $0.0570 and $0.0507 with **0 tokens** + — the write path was always wired, the poster supplied zeros. Take the counts from the same + place the METERED-SPEND LEDGER takes the dollars (codex reported usage, managed_agents + `TokensIn`/`TokensOut`, the quorum reviewer's usage block). Quota lanes still post 0/0 — that is + structural, not missing data. + The subcommand: (a) flushes any previously-spooled iterations first; (b) writes the chain + per-stage cost/tokens/model (metered) or quota bucket (encoded in `agent_id` as ` -(quota:)` — NO schema change); (c) if the observatory is unreachable, buffers to a bounded -JSONL spool (≤100 entries / 1 MiB, drop-oldest, stderr-LOUD) the next iteration flushes. It exits 0 -even on telemetry failure — a broken tracker must never wedge the loop. Review the fleet's spend -later with `ailang chains stats --by-mission` (M3). +(quota:)` — NO schema change), sets each stage's status, and **aggregates the stage +cost/tokens into the chain total** (before M2 the chain read `$0.0000` while holding $0.1077); +(c) if the observatory is unreachable, buffers to a bounded JSONL spool (≤100 entries / 1 MiB, +drop-oldest, stderr-LOUD) the next iteration flushes. It exits 0 even on telemetry failure — a +broken tracker must never wedge the loop. Review the fleet's spend later with +`ailang chains stats --by-mission` (M3). + +**Cloud leg (M3, node-generic).** When this node has a cloud observatory configured +(`AILANG_STORAGE=gcp` + `AILANG_CLOUD_PROJECT`), the same post is dual-written local **and** cloud, +so a mission iteration is queryable cloud-side alongside its OpenRouter Broadcast spans. Nothing +about it is rig-specific — a laptop or a Cloud Run job with the same env dual-writes identically, +and a node with no cloud configured behaves exactly as before. A cloud outage buffers to a separate +bounded spool and still exits 0. **GPU rule (two-tier)**: default iterations never touch `rig.lock` — it is a GPU mutex only. If (and only if) a step drives ollama/local models: `source tools/launchd/rig-lock.sh && diff --git a/changelogs/v0.18-current.md b/changelogs/v0.18-current.md index a0c56dfff..46b253d01 100644 --- a/changelogs/v0.18-current.md +++ b/changelogs/v0.18-current.md @@ -4,6 +4,46 @@ ## [Unreleased] +### Fixed — a mission iteration reported `$0.0000` while holding `$0.1077` (M-MISSION-LOOP-UNIFIED-TELEMETRY M2) + +Measured on `manual:mission:v1/iter-190`: four stages spanning Anthropic, OpenAI and OpenRouter, all +four reading `pending`, the two metered ones carrying cost with **0 tokens**, and a chain total of +`$0.0000`. Two defects with **different owners**, which is why fixing "mission accounting" as one +thing would have patched one and missed the other: + +- **Writer-side (status).** `PostIteration` called `CreateStage` → `UpdateStageMetrics` → + `UpdateStageEvalAssessment` and never `UpdateStageStatus`, so every stage kept `CreateStage`'s + `pending` default. `IterationStage` now carries a `status` field. +- **Caller-side (tokens).** `IterationStage` already carried `TokensIn`/`TokensOut` and already + passed them to `UpdateStageMetrics` — the zeros came from the poster. The mission-control skill + now supplies the real counts from the same place it takes the metered dollars. + +Stage cost and tokens are also now aggregated into the chain total, which nothing credited before. + +`status` is deliberately **per stage**: a stage that failed reads back `failed`. Blanket-completing +an iteration's stages would satisfy "no stage remains pending" while hiding the failure, so it is +tested against, not just avoided. Omitting `status` stays valid and leaves the stage `pending` — the +skill and the CLI ship independently, so an unversioned payload must keep working — while an +unrecognised value is rejected loudly rather than coerced. + +### Added — node-generic dual-write and opt-in remote reads for mission telemetry (M3) + +`ailang chains post-iteration` wrote only to the local SQLite observatory, so a mission iteration +could never be joined cloud-side with the OpenRouter Broadcast spans it produced. It now writes to +**every destination this node has configured**: local always, plus a cloud leg when +`AILANG_STORAGE=gcp`. Nothing is rig-specific — a laptop, this server or a Cloud Run job with the +same configuration behave identically, and a node with no cloud configured behaves exactly as before. + +Offline behaviour **extends the existing bounded+loud spool** rather than inventing a policy: each +leg buffers only its OWN failures (a shared spool would replay a post the local leg already stored, +duplicating that chain on every flush), the 100-entry / 1 MiB caps still hold, every buffering event +warns on stderr, and the command still exits 0 — telemetry must never wedge the loop. + +`ailang chains list` and `ailang chains view` take `--remote` to read the configured cloud +observatory. Local stays the default so offline nodes remain first-class, and `--remote` with no +cloud configured is an **error**: a remote query silently answered from SQLite would report a +cloud-side record as absent when it was never queried. + ### Fixed — `exit()` in one batch item killed the whole batch run (#607) `ailang run --batch` promises per-item isolation: it runs the entrypoint once per input, counts diff --git a/cmd/ailang/chains.go b/cmd/ailang/chains.go index 38693e62e..16c9fe692 100644 --- a/cmd/ailang/chains.go +++ b/cmd/ailang/chains.go @@ -46,6 +46,7 @@ func chainsCommand() { fmt.Println(" ailang chains tree # View as tree") fmt.Println(" ailang chains stats --hours 168 # Last week's cost summary") fmt.Println(" ailang chains stats --by-mission # Per-mission metered total vs budget + quota buckets") + fmt.Println(" ailang chains view --remote # Read this node's cloud observatory (opt-in)") fmt.Println(" ailang chains post-iteration # Post a mission iteration chain (JSON on stdin)") fmt.Println(" ailang chains diagnose # Quick issue check") fmt.Println(" ailang chains diff # Git diff across all stages") @@ -105,18 +106,19 @@ func chainsListCommand() { limit := fs.Int("limit", 20, "Maximum number of chains to show") jsonOutput := fs.Bool("json", false, "Output as JSON") fullIDs := fs.Bool("full", false, "Show full chain IDs (for copy-paste)") + remote := fs.Bool("remote", false, remoteFlagUsage) fs.Parse(flag.Args()[2:]) - // Connect to observatory database - dbPath := observatory.DefaultDatabasePath() - backend, err := observatory.NewSQLiteBackendFromPath(dbPath) + ctx := context.Background() + + // Connect to the observatory (local by default; --remote is opt-in). + backend, closeBackend, err := openChainBackend(ctx, *remote) if err != nil { fmt.Fprintf(os.Stderr, "Error: failed to connect to observatory: %v\n", err) os.Exit(1) } - defer backend.Close() + defer closeBackend() - ctx := context.Background() opts := observatory.ChainListOptions{ Limit: *limit, } @@ -190,6 +192,7 @@ func chainsViewCommand() { includeSpans := fs.Bool("spans", false, "Include span summaries for each stage (no attributes)") fullSpans := fs.Bool("full", false, "Include full span data with attributes (heavy)") jsonOutput := fs.Bool("json", false, "Output as JSON") + remote := fs.Bool("remote", false, remoteFlagUsage) fs.Parse(flag.Args()[2:]) if fs.NArg() < 1 { @@ -199,16 +202,15 @@ func chainsViewCommand() { chainIDPrefix := fs.Arg(0) - // Connect to observatory database - dbPath := observatory.DefaultDatabasePath() - backend, err := observatory.NewSQLiteBackendFromPath(dbPath) + ctx := context.Background() + + // Connect to the observatory (local by default; --remote is opt-in). + backend, closeBackend, err := openChainBackend(ctx, *remote) if err != nil { fmt.Fprintf(os.Stderr, "Error: failed to connect to observatory: %v\n", err) os.Exit(1) } - defer backend.Close() - - ctx := context.Background() + defer closeBackend() // Resolve short ID prefix to full ID chainID, err := resolveChainID(backend, ctx, chainIDPrefix) diff --git a/cmd/ailang/chains_post.go b/cmd/ailang/chains_post.go index 2c36fe68f..585b853ea 100644 --- a/cmd/ailang/chains_post.go +++ b/cmd/ailang/chains_post.go @@ -9,16 +9,26 @@ package main // drop-oldest) that the NEXT invocation flushes. It NEVER blocks or fails the // iteration — telemetry problems exit 0 with a stderr warning. // +// M-MISSION-LOOP-UNIFIED-TELEMETRY M2 added a per-stage `status` and chain-total +// aggregation; M3 made the destination NODE-GENERIC — this node dual-writes to +// local AND cloud when a cloud observatory is configured, each leg with its own +// bounded spool. +// // Input is a JSON IterationPost read from --file or stdin: // // { // "source": "mission:v1/iter-42", // "stages": [ -// {"role":"codex-executor","provider":"codex","model":"claude-sonnet-4-5","cost_usd":0.42,"tokens_in":1000,"tokens_out":500}, -// {"role":"controller","quota_bucket":"opus"}, -// {"role":"evaluator","quota_bucket":"sonnet"} +// {"role":"codex-executor","provider":"codex","model":"claude-sonnet-4-5","cost_usd":0.42,"tokens_in":1000,"tokens_out":500,"status":"completed"}, +// {"role":"controller","quota_bucket":"opus","status":"completed"}, +// {"role":"evaluator","quota_bucket":"sonnet","status":"failed"} // ] // } +// +// `status` is OPTIONAL (an older payload omitting it keeps working and leaves the +// stage pending) and is PER STAGE: a stage that failed must say `failed`, because +// blanket-completing an iteration's stages would satisfy "nothing left pending" +// while hiding the failure. import ( "context" @@ -28,8 +38,10 @@ import ( "io" "os" "path/filepath" + "strings" "github.com/sunholo-data/ailang/internal/observatory" + "github.com/sunholo-data/ailang/internal/storage" ) // defaultSpoolPath returns the mission iteration spool path next to observatory.db. @@ -41,6 +53,14 @@ func defaultSpoolPath() string { return filepath.Join(home, ".ailang", "state", "chains-iteration-spool.jsonl") } +// cloudSpoolPath derives the cloud leg's spool from the local one. The legs spool +// SEPARATELY on purpose: a shared buffer would replay a post the local leg +// already stored, duplicating that chain on every flush. +func cloudSpoolPath(localSpool string) string { + ext := filepath.Ext(localSpool) + return strings.TrimSuffix(localSpool, ext) + "-cloud" + ext +} + func chainsPostIterationCommand() { fs := flag.NewFlagSet("chains post-iteration", flag.ExitOnError) file := fs.String("file", "", "Read the iteration JSON from this file (default: stdin)") @@ -52,21 +72,14 @@ func chainsPostIterationCommand() { if spPath == "" { spPath = defaultSpoolPath() } - spool := observatory.NewSpool(spPath) - - // Connecting to the store can itself fail (locked/missing). Treat that as the - // server-down path: buffer the new post and return fail-soft. - backend, connErr := observatory.NewSQLiteBackendFromPath(observatory.DefaultDatabasePath()) - if backend != nil { - defer backend.Close() - } ctx := context.Background() - // 1. Flush any previously-spooled posts first (best-effort). - if connErr == nil { - flushSpool(ctx, backend, spool) - } + legs, closeLegs := iterationLegs(ctx, observatory.DefaultDatabasePath(), spPath) + defer closeLegs() + + // 1. Flush any previously-spooled posts first, per leg (best-effort). + observatory.FlushLegs(ctx, legs, os.Stderr) if *flushOnly { return @@ -81,36 +94,69 @@ func chainsPostIterationCommand() { return } - // 3. Try to post; on failure, spool it (LOUD, bounded, fail-soft). - if connErr != nil { - fmt.Fprintf(os.Stderr, "chains post-iteration: observatory unreachable (%v)\n", connErr) - _ = spool.Append(post) - return + // 3. Write to every leg; per-leg failures spool (LOUD, bounded, fail-soft). + // Report where the data ACTUALLY went — claiming delivery to a leg that only + // buffered would make a cloud outage invisible in the loop's own output. + delivered, spooled := observatory.PostToLegs(ctx, legs, post, os.Stderr) + if len(delivered) > 0 { + fmt.Printf("Posted iteration %s (%d stages) to %s\n", post.Source, len(post.Stages), strings.Join(delivered, "+")) } - - chainID, postErr := observatory.PostIteration(ctx, backend, post) - if postErr != nil { - fmt.Fprintf(os.Stderr, "chains post-iteration: write failed (%v)\n", postErr) - _ = spool.Append(post) - return + if len(spooled) > 0 { + fmt.Printf("Buffered iteration %s for %s (will flush next invocation)\n", post.Source, strings.Join(spooled, "+")) } - fmt.Printf("Posted iteration chain %s (source %s, %d stages)\n", chainID, post.Source, len(post.Stages)) } -// flushSpool drains buffered posts and re-posts them; posts that still fail are -// re-spooled (LOUD) so nothing is lost. -func flushSpool(ctx context.Context, backend *observatory.SQLiteBackend, spool *observatory.Spool) { - entries, err := spool.Drain() +// iterationLegs builds this NODE's dual-write destinations. The local SQLite +// observatory is always a leg (`ailang chains` is offline-first). The cloud leg +// exists only when this node is configured for one — nothing here is +// rig-specific, so a laptop, this server or a Cloud Run job with the same +// AILANG_STORAGE configuration behave identically, and a node with no cloud +// configured behaves exactly as it did before M3. +// +// A leg that cannot be OPENED is still returned, with Sink nil and Err set, so +// the post is buffered rather than dropped. +func iterationLegs(ctx context.Context, dbPath, localSpool string) ([]observatory.IterationLeg, func()) { + var closers []func() + closeAll := func() { + for _, c := range closers { + c() + } + } + + local := observatory.IterationLeg{ + Name: "local", + Spool: observatory.NewSpool(localSpool), + } + backend, err := observatory.NewSQLiteBackendFromPath(dbPath) if err != nil { - fmt.Fprintf(os.Stderr, "chains post-iteration: could not read spool (%v)\n", err) - return + local.Err = err + } else { + local.Sink = backend.Store() + closers = append(closers, func() { _ = backend.Close() }) } - for _, p := range entries { - if _, err := observatory.PostIteration(ctx, backend, p); err != nil { - fmt.Fprintf(os.Stderr, "chains post-iteration: re-post of spooled %q failed (%v); re-buffering\n", p.Source, err) - _ = spool.Append(p) - } + legs := []observatory.IterationLeg{local} + + // The cloud leg is opt-in via the SAME selector every other AILANG service + // uses (AILANG_STORAGE); adding a second selection mechanism would be one + // more thing to keep in sync. Only "gcp" names a genuinely remote + // observatory — "hybrid" still resolves the observatory to local SQLite, so + // treating it as a cloud leg would dual-write the same database twice. + if storage.GetMode() != storage.ModeGCP { + return legs, closeAll + } + + cloud := observatory.IterationLeg{ + Name: "cloud", + Spool: observatory.NewSpool(cloudSpoolPath(localSpool)), + } + backends, err := storage.NewBackends(ctx) + if err != nil { + cloud.Err = err + } else { + cloud.Sink = backends.Observatory + closers = append(closers, func() { _ = backends.Close() }) } + return append(legs, cloud), closeAll } // readIterationPost reads and decodes an IterationPost from a file or stdin. diff --git a/cmd/ailang/chains_remote.go b/cmd/ailang/chains_remote.go new file mode 100644 index 000000000..776814dfe --- /dev/null +++ b/cmd/ailang/chains_remote.go @@ -0,0 +1,48 @@ +package main + +// Opt-in remote reads for `ailang chains` (M-MISSION-LOOP-UNIFIED-TELEMETRY M3). +// +// RATIFIED (Mark, 2026-08-13): local analysis may read cloud, but OPT-IN. The +// local store stays the default so offline nodes remain first-class, and opt-in +// is the reversible choice — if `--remote` turns out to be always-passed, +// flipping the default later is one line with evidence behind it, whereas +// canonical-first is the version you cannot walk back. +// +// Consequence enforced here: `--remote` with no cloud configured is an ERROR, +// never a quiet fall back to the local store. A command that says "remote" and +// silently answers from SQLite would report a mission iteration as absent +// cloud-side when it is merely being read from the wrong place. + +import ( + "context" + "fmt" + + "github.com/sunholo-data/ailang/internal/observatory" + "github.com/sunholo-data/ailang/internal/storage" +) + +// remoteFlagUsage is the shared help text, so every command that opts in +// describes the same behaviour. +const remoteFlagUsage = "Read from this node's configured cloud observatory instead of the local store (requires AILANG_STORAGE=gcp)" + +// openChainBackend opens the observatory a chains command should READ from. +// The returned closer is always non-nil. +func openChainBackend(ctx context.Context, remote bool) (observatory.Backend, func(), error) { + if !remote { + backend, err := observatory.NewSQLiteBackendFromPath(observatory.DefaultDatabasePath()) + if err != nil { + return nil, func() {}, err + } + return backend, func() { _ = backend.Close() }, nil + } + + if mode := storage.GetMode(); mode != storage.ModeGCP { + return nil, func() {}, fmt.Errorf("--remote requires a cloud observatory: AILANG_STORAGE is %q (set AILANG_STORAGE=gcp and AILANG_CLOUD_PROJECT); refusing to answer a remote query from the local store", mode) + } + + backends, err := storage.NewBackends(ctx) + if err != nil { + return nil, func() {}, fmt.Errorf("open cloud observatory: %w", err) + } + return backends.Observatory, func() { _ = backends.Close() }, nil +} diff --git a/cmd/ailang/chains_remote_test.go b/cmd/ailang/chains_remote_test.go new file mode 100644 index 000000000..4cda26973 --- /dev/null +++ b/cmd/ailang/chains_remote_test.go @@ -0,0 +1,110 @@ +package main + +import ( + "context" + "path/filepath" + "strings" + "testing" + + "github.com/sunholo-data/ailang/internal/observatory" +) + +// legNames renders a leg set for failure messages. +func legNames(legs []observatory.IterationLeg) string { + names := make([]string, 0, len(legs)) + for _, l := range legs { + names = append(names, l.Name) + } + return strings.Join(names, "+") +} + +// M-MISSION-LOOP-UNIFIED-TELEMETRY M3 — node-generic destination selection. +// +// These assert the WIRING decisions, which is where "node-generic" is either +// true or false: which legs exist must come from configuration alone, with no +// host-specific branch anywhere in the path. + +func TestIterationLegs_NoCloudConfiguredIsLocalOnly(t *testing.T) { + dir := t.TempDir() + for _, mode := range []string{"", "local", "hybrid"} { + t.Run("AILANG_STORAGE="+mode, func(t *testing.T) { + t.Setenv("AILANG_STORAGE", mode) + + legs, closeLegs := iterationLegs(context.Background(), + filepath.Join(dir, "observatory.db"), filepath.Join(dir, "spool.jsonl")) + defer closeLegs() + + if len(legs) != 1 || legs[0].Name != "local" { + t.Fatalf("got %d leg(s) %v, want exactly the local leg — behaviour must be identical to before M3", len(legs), legNames(legs)) + } + }) + } +} + +// "hybrid" resolves the observatory to local SQLite, so treating it as a cloud +// leg would dual-write the same database twice. Covered above; called out here +// because it is the one mode whose name suggests otherwise. + +func TestIterationLegs_GCPAddsACloudLeg(t *testing.T) { + t.Setenv("AILANG_STORAGE", "gcp") + // Deliberately no AILANG_CLOUD_PROJECT: the cloud leg must still EXIST so the + // post is buffered, rather than being silently dropped because the + // destination could not be opened. + t.Setenv("AILANG_CLOUD_PROJECT", "") + + dir := t.TempDir() + legs, closeLegs := iterationLegs(context.Background(), + filepath.Join(dir, "observatory.db"), filepath.Join(dir, "spool.jsonl")) + defer closeLegs() + + if len(legs) != 2 { + t.Fatalf("got %d leg(s) %v, want local+cloud", len(legs), legNames(legs)) + } + cloud := legs[1] + if cloud.Name != "cloud" { + t.Fatalf("second leg is %q, want %q", cloud.Name, "cloud") + } + if cloud.Sink != nil || cloud.Err == nil { + t.Error("an unopenable cloud leg must carry a nil sink AND a reason, so PostToLegs spools instead of dropping") + } + if cloud.Spool.Path == legs[0].Spool.Path { + t.Error("legs share a spool file; a shared spool replays posts the other leg already stored") + } +} + +func TestCloudSpoolPath_IsDistinctAndNextToTheLocalOne(t *testing.T) { + tests := []struct{ local, want string }{ + {"/x/y/chains-iteration-spool.jsonl", "/x/y/chains-iteration-spool-cloud.jsonl"}, + {"spool.jsonl", "spool-cloud.jsonl"}, + {"/x/y/spool", "/x/y/spool-cloud"}, + } + for _, tt := range tests { + if got := cloudSpoolPath(tt.local); got != tt.want { + t.Errorf("cloudSpoolPath(%q) = %q, want %q", tt.local, got, tt.want) + } + } +} + +// TestOpenChainBackend_RemoteWithoutCloudFailsLoudly: the ratified read decision +// is opt-in remote, and an opt-in that quietly answers from the local store +// would report a cloud-side record as absent when it was never queried. +func TestOpenChainBackend_RemoteWithoutCloudFailsLoudly(t *testing.T) { + for _, mode := range []string{"", "local", "hybrid"} { + t.Run("AILANG_STORAGE="+mode, func(t *testing.T) { + t.Setenv("AILANG_STORAGE", mode) + + backend, closeBackend, err := openChainBackend(context.Background(), true) + defer closeBackend() + + if err == nil { + t.Fatal("openChainBackend(--remote) succeeded with no cloud configured; want a loud error, never a silent local fallback") + } + if backend != nil { + t.Error("openChainBackend returned a backend alongside its error") + } + if !strings.Contains(err.Error(), "AILANG_STORAGE") { + t.Errorf("error does not say how to configure a cloud observatory: %v", err) + } + }) + } +} diff --git a/cmd/ailang/chains_util.go b/cmd/ailang/chains_util.go index 2cae7a4be..b795fb43e 100644 --- a/cmd/ailang/chains_util.go +++ b/cmd/ailang/chains_util.go @@ -311,7 +311,14 @@ func displayChainDetails(backend *observatory.SQLiteBackend, ctx context.Context // resolveChainID resolves a short ID prefix to a full chain ID. // Returns error if no match or multiple matches (ambiguous prefix). -func resolveChainID(backend *observatory.SQLiteBackend, ctx context.Context, prefix string) (string, error) { +// chainLister is the only capability resolveChainID needs. Taking the narrow +// interface rather than *SQLiteBackend is what lets `--remote` reuse it against +// the cloud observatory (M-MISSION-LOOP-UNIFIED-TELEMETRY M3). +type chainLister interface { + ListChains(ctx context.Context, opts observatory.ChainListOptions) ([]*observatory.ChainSummary, error) +} + +func resolveChainID(backend chainLister, ctx context.Context, prefix string) (string, error) { // If prefix looks like a full UUID, use it directly if len(prefix) >= 36 { return prefix, nil diff --git a/cmd/ailang/help.go b/cmd/ailang/help.go index eec6bcbe2..0fcb717e0 100644 --- a/cmd/ailang/help.go +++ b/cmd/ailang/help.go @@ -185,10 +185,10 @@ func printHelp() { fmt.Printf(" %s Show streaming execution logs\n", cyan("coordinator logs")) fmt.Printf(" %s View execution chains (task→session→chat linkage)\n", cyan("chains")) fmt.Printf(" %s List all chains\n", cyan("chains list")) - fmt.Printf(" %s View chain stages and details\n", cyan("chains view")) + fmt.Printf(" %s View chain stages and details (--remote reads cloud)\n", cyan("chains view")) fmt.Printf(" %s ASCII tree with chat history\n", cyan("chains tree")) fmt.Printf(" %s Cost summary (add --by-mission for per-mission budget rollup)\n", cyan("chains stats")) - fmt.Printf(" %s Post a mission iteration chain (JSON stdin; fail-soft)\n", cyan("chains post-iteration")) + fmt.Printf(" %s Post a mission iteration chain (JSON stdin; fail-soft, dual-writes cloud when configured)\n", cyan("chains post-iteration")) fmt.Printf(" %s Quick health report for a chain\n", cyan("chains diagnose")) fmt.Printf(" %s System-wide data capture validation\n", cyan("chains health")) fmt.Printf(" %s Dashboard operations for task visualization\n", cyan("dashboard")) diff --git a/design_docs/planned/v0_33_2/HANDOVER-mission-loop-unified-telemetry.md b/design_docs/planned/v0_33_2/HANDOVER-mission-loop-unified-telemetry.md index 83f56de42..f5b1d6b75 100644 --- a/design_docs/planned/v0_33_2/HANDOVER-mission-loop-unified-telemetry.md +++ b/design_docs/planned/v0_33_2/HANDOVER-mission-loop-unified-telemetry.md @@ -9,11 +9,16 @@ Start with `.claude/skills/sprint-executor/scripts/session_start.sh M-MISSION-LO ## Status +> **SUPERSEDED 2026-08-13 — M2 and M3 are implemented.** Kept for the traps section, which is still +> accurate, and for the ratified Design Freeze wording. The one thing left is the LIVE confirmation +> M3 could not make from the sprint sandbox: post a real iteration from a node with +> `AILANG_STORAGE=gcp` and read it back cloud-side with its Broadcast spans. + | Milestone | State | |-----------|-------| | M1 session-keyed chain linkage | ✅ **PASS** — landed `56b449d01` | -| **M2 mission stage accounting** | ⬜ **PENDING — start here** | -| M3 node-generic cloud routing | ⬜ pending (depends on M2) | +| M2 mission stage accounting | ✅ **PASS** — per-stage `status`, chain-total aggregation, skill supplies tokens | +| M3 node-generic cloud routing | ✅ **PASS** (code) — dual-write via `IterationSink`, per-leg spools, `--remote` reads. ⬜ live cloud read-back not yet performed | **All three Design Freeze items are RATIFIED by Mark (2026-08-13).** Do not re-open them; they are recorded verbatim with his reasoning in the sprint JSON's `design_freeze` block: diff --git a/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry-sprint-plan.md b/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry-sprint-plan.md index b296eab35..db0c3a215 100644 --- a/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry-sprint-plan.md +++ b/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry-sprint-plan.md @@ -38,11 +38,11 @@ fail-soft behaviour from scratch. Make a Broadcast span resolve to its chain. **Tasks:** -- [ ] Register the correlation `session_id` as a `sessions` row bound to `chain_id` + `stage_id` +- [x] Register the correlation `session_id` as a `sessions` row bound to `chain_id` + `stage_id` when a mission/eval stage dispatches an OpenRouter call. -- [ ] In `convertSpan`, resolve `chain_id` via the `sessions` table when `session.id` is present and +- [x] In `convertSpan`, resolve `chain_id` via the `sessions` table when `session.id` is present and no explicit `ailang.chain_id` was supplied. -- [ ] `ailang.chain_id` keeps precedence — the new path only fires when it is absent. +- [x] `ailang.chain_id` keeps precedence — the new path only fires when it is absent. **Acceptance criteria:** - An OTLP/JSON span carrying `session.id` matching a seeded session resolves to that session's @@ -60,14 +60,14 @@ Two defects with **different owners**. A single "fix mission accounting" change miss the other, so they are separate tasks with separate criteria. **Tasks:** -- [ ] **Writer-side (status)**: `PostIteration` creates stages and never transitions them, so they +- [x] **Writer-side (status)**: `PostIteration` creates stages and never transitions them, so they keep `CreateStage`'s `StageStatusPending` default. Add a `Status` field to `IterationStage` and call `UpdateStageStatus`. Vocabulary available: `pending`, `running`, `awaiting_approval`, `completed`, `failed`. -- [ ] **Caller-side (tokens)**: `IterationStage` ALREADY carries `TokensIn`/`TokensOut`; +- [x] **Caller-side (tokens)**: `IterationStage` ALREADY carries `TokensIn`/`TokensOut`; `UpdateStageMetrics` already receives them. The zeros come from the poster. Supply real token counts from the mission-control skill. -- [ ] Aggregate stage cost/tokens into the chain total. +- [x] Aggregate stage cost/tokens into the chain total. **Acceptance criteria:** - A posted stage with a terminal status reads back as that status, NOT `pending`. @@ -82,11 +82,11 @@ miss the other, so they are separate tasks with separate criteria. ### M3_NODE_GENERIC_CLOUD_ROUTING (~120 LOC, ~4h) **Tasks:** -- [ ] Make the backend at `chains_post.go:59` selectable rather than hardcoded SQLite, reusing +- [x] Make the backend at `chains_post.go:59` selectable rather than hardcoded SQLite, reusing `internal/storage.NewBackends` which already resolves local/gcp/hybrid from `AILANG_STORAGE`. -- [ ] Dual-write: local AND cloud, per the ratified decision. -- [ ] Confirm the EXISTING spool covers a cloud write failure — extend, do not replace. -- [ ] Opt-in remote read for analysis; local stays the default. +- [x] Dual-write: local AND cloud, per the ratified decision. +- [x] Confirm the EXISTING spool covers a cloud write failure — extend, do not replace. +- [x] Opt-in remote read for analysis; local stays the default. **Acceptance criteria:** - With cloud configured, a posted iteration appears in BOTH stores. diff --git a/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry.md b/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry.md index ccbb741b0..d586be8b1 100644 --- a/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry.md +++ b/design_docs/planned/v0_33_2/m-mission-loop-unified-telemetry.md @@ -152,12 +152,12 @@ no new schema, no second model. ### Implementation Plan **M1: Session-keyed chain linkage** (~6 hours) -- [ ] Register the correlation `session_id` as a `sessions` row bound to `chain_id` + `stage_id` when +- [x] Register the correlation `session_id` as a `sessions` row bound to `chain_id` + `stage_id` when a mission/eval stage dispatches an OpenRouter call. -- [ ] Extend `convertSpan` to resolve `chain_id` via `sessions` when `session.id` is present and no +- [x] Extend `convertSpan` to resolve `chain_id` via `sessions` when `session.id` is present and no explicit `ailang.chain_id` was supplied — **without** disturbing the Claude Code path that owns that attribute today. -- [ ] Test both directions: a Claude Code span still resolves as it does now; an OpenRouter Broadcast +- [x] Test both directions: a Claude Code span still resolves as it does now; an OpenRouter Broadcast span resolves to its chain. **M2: Mission stage lifecycle + accounting** (~8 hours) @@ -165,19 +165,22 @@ no new schema, no second model. Two *distinct* defects with different owners — worth separating, because a single "fix mission accounting" change would likely patch one and miss the other: -- [ ] **Status (writer-side).** `PostIteration` creates stages and never transitions them, so they +- [x] **Status (writer-side).** `PostIteration` creates stages and never transitions them, so they keep `CreateStage`'s `StageStatusPending` default. Add the status transition in `iteration_post.go`, carrying per-stage outcome rather than blanket-completing. -- [ ] **Tokens (caller-side).** `UpdateStageMetrics` already receives `st.TokensIn`/`st.TokensOut` +- [x] **Tokens (caller-side).** `UpdateStageMetrics` already receives `st.TokensIn`/`st.TokensOut` correctly; the zeros come from the caller. Thread real token counts through `ailang chains post-iteration` and the mission-control skill that invokes it. -- [ ] Aggregate stage cost/tokens into the chain total. -- [ ] Regression fixture built from the real iter-190 shape (4 stages, 3 providers, cost-without-tokens). +- [x] Aggregate stage cost/tokens into the chain total. +- [x] Regression fixture built from the real iter-190 shape (4 stages, 3 providers, cost-without-tokens). **M3: Cloud routing for rig writes** (~8 hours) -- [ ] Wire the rig's observatory writes to the Firestore backend per the frozen decision. -- [ ] Offline behaviour per the frozen decision. +- [x] Wire the rig's observatory writes to the Firestore backend per the frozen decision. +- [x] Offline behaviour per the frozen decision. - [ ] Verify a full mission iteration lands cloud-side with every stage and its Broadcast spans. + **NOT DONE — needs a live run on a `AILANG_STORAGE=gcp` node.** The write path and its + spool are implemented and unit-tested against a non-SQLite sink; the end-to-end confirmation + is a manual step that cannot be performed from the sprint sandbox (no cloud credentials). ### Files to Modify/Create @@ -214,13 +217,17 @@ ailang chains view --remote ## Success Criteria -- [ ] One query returns a full mission iteration across all four providers -- [ ] Chain total equals the sum of stage costs (regression fixture from iter-190) -- [ ] ≥95% of a mission iteration's Broadcast spans carry a resolved `chain_id` -- [ ] No stage ends an iteration in `pending` -- [ ] **Claude Code session correlation is unchanged** — asserted by a test that fails if the +- [ ] One query returns a full mission iteration across all four providers — **needs a live + mission iteration**; the code path exists and is unit-tested, the measurement is not made +- [x] Chain total equals the sum of stage costs (regression fixture from iter-190) +- [ ] ≥95% of a mission iteration's Broadcast spans carry a resolved `chain_id` — **M1 landed and + is unit-tested; the ≥95% figure is a live measurement, not yet taken** +- [x] No stage ends an iteration in `pending` — the CLI accepts and stores a per-stage status and + the mission-control skill now supplies it; a stage that omits one is still left `pending` by + design (version skew), so the live figure depends on the skill having been updated +- [x] **Claude Code session correlation is unchanged** — asserted by a test that fails if the existing path regresses -- [ ] All tests passing; `make lint` clean +- [x] All tests passing; `make lint` clean ## Testing Strategy diff --git a/internal/observatory/iteration_legs.go b/internal/observatory/iteration_legs.go new file mode 100644 index 000000000..d0beaef38 --- /dev/null +++ b/internal/observatory/iteration_legs.go @@ -0,0 +1,118 @@ +package observatory + +import ( + "context" + "fmt" + "io" + "os" +) + +// Node-generic dual-write for mission iterations +// (M-MISSION-LOOP-UNIFIED-TELEMETRY M3). +// +// RATIFIED (Mark, 2026-08-13): dual-write rather than mirror, and deliberately +// NODE-GENERIC — "this server, laptop, cloud, other nodes in the future". +// Nothing here knows about "the rig": a leg is just a sink plus its own spool, +// and WHICH legs exist is decided by the node's configuration at the call site +// (cmd/ailang/chains_post.go), not by any host-specific test in this package. +// +// RATIFIED: never block when a destination is unreachable — "no block if not +// available, at least until we harden availability." That is already satisfied +// structurally by the bounded+loud spool (spool.go), so this EXTENDS the spool +// to a second destination rather than inventing a fail-soft policy. Availability +// hardening is deferred, not forgotten; the spool's bounded+loud contract is +// what keeps "no block" honest rather than silently lossy. +// +// PER-LEG SPOOLS ARE LOAD-BEARING. A shared spool would replay a post that the +// local leg already stored, duplicating that chain on every subsequent flush. +// Each leg therefore buffers only its OWN failures. + +// The local SQLite store is the always-present leg, and the only one that can +// record a stage's model. +var ( + _ IterationSink = (*Store)(nil) + _ IterationModelSink = (*Store)(nil) +) + +// IterationLeg is one destination of a dual-write, with the bounded spool that +// covers its outages. +type IterationLeg struct { + // Name identifies the leg in stderr notices ("local", "cloud", …). + Name string + // Sink is the destination. Nil means the destination could not be opened — + // see Err. A nil sink still spools; it does not drop. + Sink IterationSink + // Err records why Sink is nil (connect-time failure). + Err error + // Spool buffers THIS leg's failures only. + Spool *Spool +} + +// warnTo resolves the loud-notice sink (nil => stderr, matching Spool). +func warnTo(w io.Writer) io.Writer { + if w == nil { + return os.Stderr + } + return w +} + +// PostToLegs writes post to every leg, buffering per-leg failures to that leg's +// spool. It never fails: telemetry must not block or wedge a mission iteration. +// Every buffering event is loud. +// +// It returns which legs STORED the post and which only BUFFERED it, so the +// caller reports where the data actually went rather than claiming delivery it +// did not get. +// +// A post that fails VALIDATION is a caller bug, not an outage, so it is reported +// and dropped rather than spooled — buffering it would replay the same rejection +// on every future iteration and evict recoverable posts from the bounded spool. +func PostToLegs(ctx context.Context, legs []IterationLeg, post *IterationPost, warn io.Writer) (delivered, spooled []string) { + w := warnTo(warn) + + if err := post.Validate(); err != nil { + fmt.Fprintf(w, "chains post-iteration: invalid post %q not buffered (%v)\n", post.Source, err) + return nil, nil + } + + for _, leg := range legs { + if leg.Sink == nil { + fmt.Fprintf(w, "chains post-iteration: %s observatory unreachable (%v)\n", leg.Name, leg.Err) + _ = leg.Spool.Append(post) + spooled = append(spooled, leg.Name) + continue + } + if _, err := PostIterationTo(ctx, leg.Sink, post); err != nil { + fmt.Fprintf(w, "chains post-iteration: %s write failed (%v)\n", leg.Name, err) + _ = leg.Spool.Append(post) + spooled = append(spooled, leg.Name) + continue + } + delivered = append(delivered, leg.Name) + } + return delivered, spooled +} + +// FlushLegs drains each leg's spool and replays it against that leg. A replay +// that still fails is re-buffered (loudly) so nothing is lost. A leg whose sink +// could not be opened is skipped with its backlog intact. +func FlushLegs(ctx context.Context, legs []IterationLeg, warn io.Writer) { + w := warnTo(warn) + + for _, leg := range legs { + if leg.Sink == nil { + continue // still down; leave the backlog where it is + } + entries, err := leg.Spool.Drain() + if err != nil { + fmt.Fprintf(w, "chains post-iteration: could not read %s spool (%v)\n", leg.Name, err) + continue + } + for _, p := range entries { + if _, err := PostIterationTo(ctx, leg.Sink, p); err != nil { + fmt.Fprintf(w, "chains post-iteration: %s re-post of spooled %q failed (%v); re-buffering\n", leg.Name, p.Source, err) + _ = leg.Spool.Append(p) + } + } + } +} diff --git a/internal/observatory/iteration_legs_test.go b/internal/observatory/iteration_legs_test.go new file mode 100644 index 000000000..8bf7f4fa1 --- /dev/null +++ b/internal/observatory/iteration_legs_test.go @@ -0,0 +1,239 @@ +package observatory + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" +) + +// M-MISSION-LOOP-UNIFIED-TELEMETRY M3 — node-generic dual-write. +// +// The ratified decision is dual-write, NODE-GENERIC ("this server, laptop, cloud, +// other nodes in the future") and NEVER-BLOCK when a destination is unreachable. +// Never-block is already satisfied structurally by the bounded+loud spool, so +// these tests assert the EXTENSION of that spool to a second leg rather than a +// new fail-soft policy. +// +// The subtle requirement: each leg spools SEPARATELY. A shared spool would replay +// a post that the local leg already stored, duplicating the chain on every flush. + +func testLeg(t *testing.T, name string, sink IterationSink, warn *bytes.Buffer) IterationLeg { + t.Helper() + sp := NewSpool(filepath.Join(t.TempDir(), name+"-spool.jsonl")) + sp.SetWarnWriter(warn) + return IterationLeg{Name: name, Sink: sink, Spool: sp} +} + +func TestPostToLegs_DualWriteReachesBothStores(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + cloud := newFakeSink() // Firestore: no eval_assessment support + + legs := []IterationLeg{ + testLeg(t, "local", local, &warn), + testLeg(t, "cloud", cloud, &warn), + } + PostToLegs(context.Background(), legs, iter190Post(), &warn) + + for _, sink := range []*fakeSink{local.fakeSink, cloud} { + if len(sink.chains) != 1 { + t.Errorf("sink wrote %d chains, want 1", len(sink.chains)) + } + if len(sink.stages) != 4 { + t.Errorf("sink wrote %d stages, want 4", len(sink.stages)) + } + } + for _, leg := range legs { + if n := leg.Spool.Len(); n != 0 { + t.Errorf("leg %q spooled %d posts after a successful write, want 0", leg.Name, n) + } + } +} + +// TestPostToLegs_CloudUnreachableNeverBlocks is the ratified never-block +// requirement, asserted against a deliberately-broken cloud leg. +func TestPostToLegs_CloudUnreachableNeverBlocks(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + cloud := newFakeSink() + cloud.failOn = "CreateChain" // the store is there but refusing writes + + legs := []IterationLeg{ + testLeg(t, "local", local, &warn), + testLeg(t, "cloud", cloud, &warn), + } + PostToLegs(context.Background(), legs, iter190Post(), &warn) + + if len(local.chains) != 1 { + t.Errorf("local leg wrote %d chains, want 1 — a cloud outage must not stop the local write", len(local.chains)) + } + if n := legs[1].Spool.Len(); n != 1 { + t.Errorf("cloud spool holds %d posts, want 1", n) + } + // The local leg succeeded, so it must NOT be replayed later — that would + // duplicate the chain it already stored. + if n := legs[0].Spool.Len(); n != 0 { + t.Errorf("local spool holds %d posts, want 0 (it succeeded)", n) + } + if got := warn.String(); !strings.Contains(got, "cloud") { + t.Errorf("no loud notice naming the failed leg; got:\n%s", got) + } +} + +// TestPostToLegs_ReportsWhereTheDataWent: the caller prints this, so claiming +// delivery to a leg that only buffered would make a cloud outage invisible in +// the loop's own output. +func TestPostToLegs_ReportsWhereTheDataWent(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + cloud := newFakeSink() + cloud.failOn = "CreateChain" + + legs := []IterationLeg{ + testLeg(t, "local", local, &warn), + testLeg(t, "cloud", cloud, &warn), + } + delivered, spooled := PostToLegs(context.Background(), legs, iter190Post(), &warn) + + if len(delivered) != 1 || delivered[0] != "local" { + t.Errorf("delivered = %v, want [local]", delivered) + } + if len(spooled) != 1 || spooled[0] != "cloud" { + t.Errorf("spooled = %v, want [cloud]", spooled) + } + + // An invalid post is neither delivered nor buffered. + delivered, spooled = PostToLegs(context.Background(), legs, &IterationPost{Source: "x"}, &warn) + if len(delivered) != 0 || len(spooled) != 0 { + t.Errorf("invalid post reported delivered=%v spooled=%v, want both empty", delivered, spooled) + } +} + +// TestPostToLegs_UnavailableLegSpoolsWithoutASink covers the connect-time +// failure: the destination could not be opened at all, so there is no sink to +// call. The post must still be buffered rather than dropped. +func TestPostToLegs_UnavailableLegSpoolsWithoutASink(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + cloud := testLeg(t, "cloud", nil, &warn) + cloud.Err = context.DeadlineExceeded + + legs := []IterationLeg{testLeg(t, "local", local, &warn), cloud} + PostToLegs(context.Background(), legs, iter190Post(), &warn) + + if n := cloud.Spool.Len(); n != 1 { + t.Errorf("cloud spool holds %d posts, want 1 (an unopenable leg still buffers)", n) + } + if len(local.chains) != 1 { + t.Errorf("local leg wrote %d chains, want 1", len(local.chains)) + } +} + +// TestPostToLegs_NoCloudLegIsIdenticalToToday: with no cloud configured the +// caller passes one leg, and nothing about the local path changes — in +// particular no second spool file is created. +func TestPostToLegs_NoCloudLegIsIdenticalToToday(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + leg := testLeg(t, "local", local, &warn) + + PostToLegs(context.Background(), []IterationLeg{leg}, iter190Post(), &warn) + + if len(local.chains) != 1 { + t.Fatalf("local leg wrote %d chains, want 1", len(local.chains)) + } + entries, err := os.ReadDir(filepath.Dir(leg.Spool.Path)) + if err != nil { + t.Fatalf("read spool dir: %v", err) + } + if len(entries) != 0 { + t.Errorf("single-leg post created %d file(s) in the spool dir, want 0: %v", len(entries), entries) + } + if warn.Len() != 0 { + t.Errorf("single-leg success warned on stderr: %s", warn.String()) + } +} + +// TestPostToLegs_CloudSpoolStaysBounded: adding a second leg must not weaken the +// spool's existing caps. +func TestPostToLegs_CloudSpoolStaysBounded(t *testing.T) { + var warn bytes.Buffer + cloud := newFakeSink() + cloud.failOn = "CreateChain" + leg := testLeg(t, "cloud", cloud, &warn) + + for i := 0; i < DefaultSpoolMaxEntries+25; i++ { + PostToLegs(context.Background(), []IterationLeg{leg}, iter190Post(), &warn) + } + if n := leg.Spool.Len(); n != DefaultSpoolMaxEntries { + t.Errorf("cloud spool holds %d posts, want the %d-entry cap", n, DefaultSpoolMaxEntries) + } + if !strings.Contains(warn.String(), "OVERFLOW") { + t.Error("spool overflowed without a loud OVERFLOW notice") + } +} + +func TestFlushLegs_ReplaysOnlyItsOwnSpool(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + cloud := newFakeSink() + cloud.failOn = "CreateChain" + + legs := []IterationLeg{ + testLeg(t, "local", local, &warn), + testLeg(t, "cloud", cloud, &warn), + } + PostToLegs(context.Background(), legs, iter190Post(), &warn) + + // Cloud recovers; the next invocation flushes ONLY the cloud backlog. + cloud.failOn = "" + FlushLegs(context.Background(), legs, &warn) + + if len(cloud.chains) != 1 { + t.Errorf("cloud holds %d chains after flush, want 1", len(cloud.chains)) + } + if len(local.chains) != 1 { + t.Errorf("local holds %d chains, want 1 — a cloud flush must not re-post the local leg", len(local.chains)) + } + if n := legs[1].Spool.Len(); n != 0 { + t.Errorf("cloud spool holds %d posts after a successful flush, want 0", n) + } +} + +func TestFlushLegs_StillBrokenLegRebuffers(t *testing.T) { + var warn bytes.Buffer + cloud := newFakeSink() + cloud.failOn = "CreateChain" + legs := []IterationLeg{testLeg(t, "cloud", cloud, &warn)} + + PostToLegs(context.Background(), legs, iter190Post(), &warn) + FlushLegs(context.Background(), legs, &warn) // still broken + + if n := legs[0].Spool.Len(); n != 1 { + t.Errorf("cloud spool holds %d posts, want 1 (a failed replay must re-buffer, not drop)", n) + } +} + +// TestPostToLegs_InvalidPostIsNotSpooled: a malformed post is a caller bug, not +// an outage. Buffering it would replay the same rejection on every future +// iteration until it aged out of the cap, evicting recoverable posts. +func TestPostToLegs_InvalidPostIsNotSpooled(t *testing.T) { + var warn bytes.Buffer + local := &modelFakeSink{newFakeSink()} + leg := testLeg(t, "local", local, &warn) + + PostToLegs(context.Background(), []IterationLeg{leg}, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{{Role: "controller", Status: "finished"}}, + }, &warn) + + if n := leg.Spool.Len(); n != 0 { + t.Errorf("spool holds %d posts, want 0 (an invalid post must not be retried forever)", n) + } + if !strings.Contains(warn.String(), "invalid") { + t.Errorf("invalid post rejected without a loud notice; got:\n%s", warn.String()) + } +} diff --git a/internal/observatory/iteration_post.go b/internal/observatory/iteration_post.go index b62aeaf06..0972ef847 100644 --- a/internal/observatory/iteration_post.go +++ b/internal/observatory/iteration_post.go @@ -3,6 +3,7 @@ package observatory import ( "context" "fmt" + "os" "time" ) @@ -40,6 +41,31 @@ type IterationStage struct { // It is encoded into the free-text agent_id as " (quota:)" so it // is visible in `chains view` with NO schema change. QuotaBucket string `json:"quota_bucket,omitempty"` + // Status is the stage's OWN outcome (M-MISSION-LOOP-UNIFIED-TELEMETRY M2). + // One of the ChainStageStatus vocabulary: pending, running, awaiting_approval, + // completed, failed. + // + // It is deliberately PER-STAGE and never defaulted to "completed": blanket- + // completing an iteration's stages would satisfy "no stage remains pending" + // while hiding a stage that genuinely failed. + // + // EMPTY IS VALID and means "not supplied" — the stage keeps CreateStage's + // pending default, which is exactly today's behaviour. The mission-control + // skill and this CLI ship independently, so a payload written before Status + // existed must keep working rather than have an outcome invented for it. + Status string `json:"status,omitempty"` +} + +// validStageStatuses is the accepted Status vocabulary. An unrecognised value is +// REJECTED rather than coerced: a status is an outcome claim, and silently +// mapping an unknown one onto "completed" is the same failure-hiding this +// milestone exists to prevent. +var validStageStatuses = map[string]ChainStageStatus{ + string(StageStatusPending): StageStatusPending, + string(StageStatusRunning): StageStatusRunning, + string(StageStatusAwaitingApproval): StageStatusAwaitingApproval, + string(StageStatusCompleted): StageStatusCompleted, + string(StageStatusFailed): StageStatusFailed, } // IterationPost is one mission iteration to be posted as a chain. @@ -76,21 +102,54 @@ func (p *IterationPost) Validate() error { if st.QuotaBucket != "" && (st.TokensIn != 0 || st.TokensOut != 0 || st.CostUSD != 0) { return fmt.Errorf("iteration post: quota-lane stage %q must have zero tokens and cost (subscription spend is bucket-visible, not dollar-faked)", st.Role) } + if st.Status != "" { + if _, ok := validStageStatuses[st.Status]; !ok { + return fmt.Errorf("iteration post: stage %q has unknown status %q (valid: pending, running, awaiting_approval, completed, failed; omit for today's default)", st.Role, st.Status) + } + } } return nil } -// PostIteration writes one iteration chain and its stages to the observatory SQLite -// store. The chain and each stage are created; a per-stage failure aborts and -// returns an error (the caller spools the WHOLE post for retry). The quota bucket is -// encoded into agent_id. It takes *SQLiteBackend directly because it writes the -// stage model via eval_assessment (a store-level write not on the Backend interface) -// and because `ailang chains` is offline-first (direct SQLite). +// IterationSink is the narrow write surface an iteration post needs +// (M-MISSION-LOOP-UNIFIED-TELEMETRY M3). Every method on it is already part of +// observatory.Backend, so the local SQLite Store AND the Firestore +// ObservatoryStore both satisfy it — which is what makes the write path +// NODE-GENERIC: the node picks a sink, the poster does not know which node it is +// running on. +type IterationSink interface { + CreateChain(ctx context.Context, req *ChainCreateRequest) (*ExecutionChain, error) + CreateStage(ctx context.Context, req *StageCreateRequest) (*ChainStage, error) + UpdateStageMetrics(ctx context.Context, stageID string, cost float64, tokensIn, tokensOut, turns, toolCalls int, durationMs int64, costProvenance string) error + UpdateStageStatus(ctx context.Context, stageID string, status ChainStageStatus) error + UpdateChainMetrics(ctx context.Context, id string, cost float64, tokens, turns int) error +} + +// IterationModelSink is the OPTIONAL extension for recording a stage's model. +// UpdateStageEvalAssessment is a *Store method and is deliberately NOT on the +// Backend interface, so the Firestore observatory does not implement it. A sink +// that cannot record models is not an error — the local leg of a dual-write +// still records them, so the datum is not lost system-wide — but it is never +// silent either: PostIterationTo says so on stderr. +type IterationModelSink interface { + UpdateStageEvalAssessment(ctx context.Context, stageID string, assessment *EvalAssessment) error +} + +// PostIteration writes one iteration chain and its stages to the LOCAL observatory +// SQLite store. It is the offline-first default (`ailang chains` reads SQLite +// directly); PostIterationTo is the same write against any sink. func PostIteration(ctx context.Context, backend *SQLiteBackend, p *IterationPost) (string, error) { + return PostIterationTo(ctx, backend.Store(), p) +} + +// PostIterationTo writes one iteration chain and its stages to sink. The chain and +// each stage are created; a per-stage failure aborts and returns an error (the +// caller spools the WHOLE post for retry). The quota bucket is encoded into +// agent_id. +func PostIterationTo(ctx context.Context, store IterationSink, p *IterationPost) (string, error) { if err := p.Validate(); err != nil { return "", err } - store := backend.Store() chain, err := store.CreateChain(ctx, &ChainCreateRequest{ SourceType: ChainSourceManual, @@ -100,6 +159,17 @@ func PostIteration(ctx context.Context, backend *SQLiteBackend, p *IterationPost return "", fmt.Errorf("create iteration chain: %w", err) } + // Chain totals are denormalized counters that nothing else credits for a + // mission iteration, which is why iter-190 read $0.0000 while its stages held + // $0.1077. Accumulate here and post ONE UpdateChainMetrics after the stages. + var totalCost float64 + var totalTokens int + + // A sink that cannot record models (Firestore) drops them; count and report + // rather than dropping quietly. + modelSink, canRecordModel := store.(IterationModelSink) + modelsDropped := 0 + for i, st := range p.Stages { stage, err := store.CreateStage(ctx, &StageCreateRequest{ ChainID: chain.ID, @@ -118,13 +188,42 @@ func PostIteration(ctx context.Context, backend *SQLiteBackend, p *IterationPost } // Record the model (metered lanes) so the M1 classifier can resolve a rate. if st.Model != "" { - if err := store.UpdateStageEvalAssessment(ctx, stage.ID, &EvalAssessment{ - Model: st.Model, - EvalMode: "mission", - }); err != nil { - return chain.ID, fmt.Errorf("update stage %d model: %w", i, err) + switch { + case canRecordModel: + if err := modelSink.UpdateStageEvalAssessment(ctx, stage.ID, &EvalAssessment{ + Model: st.Model, + EvalMode: "mission", + }); err != nil { + return chain.ID, fmt.Errorf("update stage %d model: %w", i, err) + } + default: + modelsDropped++ } } + // Status LAST, so completed_at lands after the stage is fully credited. + // Only when supplied — see IterationStage.Status on why an absent status + // is left pending rather than assumed complete. + if st.Status != "" { + if err := store.UpdateStageStatus(ctx, stage.ID, validStageStatuses[st.Status]); err != nil { + return chain.ID, fmt.Errorf("update stage %d status: %w", i, err) + } + } + + totalCost += st.CostUSD + totalTokens += st.TokensIn + st.TokensOut + } + + // Aggregate the stages into the chain total. Turns are not modelled per + // iteration stage, so 0 is the honest value rather than a guess. + if totalCost != 0 || totalTokens != 0 { + if err := store.UpdateChainMetrics(ctx, chain.ID, totalCost, totalTokens, 0); err != nil { + return chain.ID, fmt.Errorf("aggregate chain metrics: %w", err) + } + } + + if modelsDropped > 0 { + fmt.Fprintf(os.Stderr, "chains post-iteration: %d stage model(s) not recorded on this sink (%T does not implement eval_assessment); cost rate resolution for %s relies on the local leg\n", + modelsDropped, store, p.Source) } return chain.ID, nil diff --git a/internal/observatory/iteration_post_test.go b/internal/observatory/iteration_post_test.go new file mode 100644 index 000000000..5653ef0f7 --- /dev/null +++ b/internal/observatory/iteration_post_test.go @@ -0,0 +1,234 @@ +package observatory + +import ( + "context" + "math" + "testing" +) + +// M-MISSION-LOOP-UNIFIED-TELEMETRY M2 — mission stage accounting. +// +// Two defects with DIFFERENT owners, which is why they get separate tests: +// +// - writer-side: PostIteration created stages and never transitioned them, so +// every stage kept CreateStage's StageStatusPending default; +// - aggregation: the chain total was never credited, so iter-190 read $0.0000 +// while holding $0.1077 of stage cost. +// +// The load-bearing test here is TestPostIteration_FailedStageReadsBackFailed: +// blanket-completing every stage would satisfy "no stage remains pending" AND +// hide real failures. That is an acceptance criterion, not a preference. + +// postToMemory posts p to a fresh in-memory observatory and returns the chain id +// plus the store, so tests assert on what was READ BACK rather than on what the +// poster believed it wrote. +func postToMemory(t *testing.T, p *IterationPost) (string, *Store) { + t.Helper() + backend, err := NewSQLiteBackendFromPath(":memory:") + if err != nil { + t.Fatalf("open in-memory observatory: %v", err) + } + t.Cleanup(func() { _ = backend.Close() }) + + chainID, err := PostIteration(context.Background(), backend, p) + if err != nil { + t.Fatalf("PostIteration: %v", err) + } + return chainID, backend.Store() +} + +// stagesByRole reads the chain's stages back and indexes them by agent_id. +func stagesByRole(t *testing.T, store *Store, chainID string) map[string]*ChainStage { + t.Helper() + stages, err := store.GetChainStages(context.Background(), chainID, ChainReadOptions{}) + if err != nil { + t.Fatalf("GetChainStages: %v", err) + } + out := make(map[string]*ChainStage, len(stages)) + for _, st := range stages { + out[st.AgentID] = st + } + return out +} + +func TestPostIteration_TerminalStatusReadsBack(t *testing.T) { + chainID, store := postToMemory(t, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{ + {Role: "controller", Provider: "anthropic", QuotaBucket: "opus", Status: "completed"}, + {Role: "quorum-r1", Provider: "openrouter", Model: "gpt-5.6-sol", CostUSD: 0.0570, Status: "completed"}, + }, + }) + + for _, agentID := range []string{"controller (quota:opus)", "quorum-r1"} { + st, ok := stagesByRole(t, store, chainID)[agentID] + if !ok { + t.Fatalf("stage %q not found", agentID) + } + if st.Status != StageStatusCompleted { + t.Errorf("stage %q status = %q, want %q (a posted terminal status must not read back pending)", + agentID, st.Status, StageStatusCompleted) + } + } +} + +// TestPostIteration_FailedStageReadsBackFailed is the criterion that blocks the +// shortcut: setting every stage to completed would satisfy "no stage remains +// pending" while hiding real failures. +func TestPostIteration_FailedStageReadsBackFailed(t *testing.T) { + chainID, store := postToMemory(t, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{ + {Role: "controller", QuotaBucket: "opus", Status: "completed"}, + {Role: "designer", QuotaBucket: "codex", Status: "failed"}, + {Role: "quorum-r1", Provider: "openrouter", CostUSD: 0.0570, Status: "completed"}, + }, + }) + + got := stagesByRole(t, store, chainID) + if st := got["designer (quota:codex)"]; st == nil || st.Status != StageStatusFailed { + var have ChainStageStatus + if st != nil { + have = st.Status + } + t.Errorf("failed stage status = %q, want %q — a failed stage must still say so", have, StageStatusFailed) + } + // And the surrounding stages keep their own outcome (no blanket transition). + if st := got["controller (quota:opus)"]; st == nil || st.Status != StageStatusCompleted { + t.Errorf("controller status = %v, want completed", st) + } + + // stages_completed counts the completed stages only, not the failed one. + chain, err := store.GetChain(context.Background(), chainID, ChainReadOptions{}) + if err != nil { + t.Fatalf("GetChain: %v", err) + } + if chain.StagesCompleted != 2 { + t.Errorf("chain.StagesCompleted = %d, want 2 (the failed stage must not count as completed)", chain.StagesCompleted) + } +} + +func TestPostIteration_TokensReadBack(t *testing.T) { + chainID, store := postToMemory(t, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{ + {Role: "executor", Provider: "codex", Model: "gpt-5.6-sol", CostUSD: 0.42, TokensIn: 18000, TokensOut: 2000, Status: "completed"}, + }, + }) + + st := stagesByRole(t, store, chainID)["executor"] + if st == nil { + t.Fatal("executor stage not found") + } + if st.TokensIn != 18000 || st.TokensOut != 2000 { + t.Errorf("stage tokens = %d in / %d out, want 18000 / 2000", st.TokensIn, st.TokensOut) + } + if math.Abs(st.Cost-0.42) > 1e-9 { + t.Errorf("stage cost = %v, want 0.42", st.Cost) + } +} + +// TestPostIteration_ChainTotalEqualsSumOfStages is the iter-190 regression +// fixture: 4 stages, 3 providers, and two stages carrying cost with ZERO tokens +// (the real measured shape). The chain reported $0.0000 while its stages held +// $0.1077. +func TestPostIteration_ChainTotalEqualsSumOfStages(t *testing.T) { + stages := []IterationStage{ + {Role: "controller", Provider: "anthropic", QuotaBucket: "opus", Status: "completed"}, + {Role: "designer", Provider: "codex", QuotaBucket: "codex", Status: "completed"}, + {Role: "quorum-r1", Provider: "openrouter", Model: "gpt-5.6-sol", CostUSD: 0.0570, Status: "completed"}, + {Role: "quorum-r2", Provider: "openrouter", Model: "gemini-3.1-pro", CostUSD: 0.0507, TokensIn: 9000, TokensOut: 1000, Status: "completed"}, + } + chainID, store := postToMemory(t, &IterationPost{Source: "manual:mission:v1/iter-190", Stages: stages}) + + var wantCost float64 + var wantTokens int + for _, st := range stages { + wantCost += st.CostUSD + wantTokens += st.TokensIn + st.TokensOut + } + + chain, err := store.GetChain(context.Background(), chainID, ChainReadOptions{}) + if err != nil { + t.Fatalf("GetChain: %v", err) + } + if math.Abs(chain.TotalCost-wantCost) > 1e-9 { + t.Errorf("chain.TotalCost = %v, want %v (sum of stage costs)", chain.TotalCost, wantCost) + } + if chain.TotalTokens != wantTokens { + t.Errorf("chain.TotalTokens = %d, want %d (sum of stage tokens)", chain.TotalTokens, wantTokens) + } + + // Cross-check against what the stages themselves read back, so the assertion + // is "total == sum of stored stages", not "total == sum of my fixture". + var storedCost float64 + var storedTokens int + for _, st := range stagesByRole(t, store, chainID) { + storedCost += st.Cost + storedTokens += st.TokensIn + st.TokensOut + } + if math.Abs(chain.TotalCost-storedCost) > 1e-9 || chain.TotalTokens != storedTokens { + t.Errorf("chain total (%v, %d) != sum of stored stages (%v, %d)", + chain.TotalCost, chain.TotalTokens, storedCost, storedTokens) + } +} + +// TestPostIteration_OmittedStatusKeepsTodaysBehaviour covers version skew: the +// mission-control skill and the CLI ship independently, so a payload written +// before Status existed must keep working and must not be silently invented. +func TestPostIteration_OmittedStatusKeepsTodaysBehaviour(t *testing.T) { + chainID, store := postToMemory(t, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{ + {Role: "controller", QuotaBucket: "opus"}, + {Role: "quorum-r1", Provider: "openrouter", CostUSD: 0.0570}, + }, + }) + + for agentID, st := range stagesByRole(t, store, chainID) { + if st.Status != StageStatusPending { + t.Errorf("stage %q status = %q, want %q — an omitted status must not be invented", + agentID, st.Status, StageStatusPending) + } + } + // Aggregation still happens: it does not depend on status being supplied. + chain, err := store.GetChain(context.Background(), chainID, ChainReadOptions{}) + if err != nil { + t.Fatalf("GetChain: %v", err) + } + if math.Abs(chain.TotalCost-0.0570) > 1e-9 { + t.Errorf("chain.TotalCost = %v, want 0.0570", chain.TotalCost) + } +} + +func TestIterationPost_ValidateStatus(t *testing.T) { + tests := []struct { + name string + status string + wantErr bool + }{ + {"empty is allowed (version skew)", "", false}, + {"pending", "pending", false}, + {"running", "running", false}, + {"awaiting_approval", "awaiting_approval", false}, + {"completed", "completed", false}, + {"failed", "failed", false}, + {"unknown status is rejected loudly", "finished", true}, + {"case mismatch is rejected loudly", "COMPLETED", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{{Role: "controller", QuotaBucket: "opus", Status: tt.status}}, + } + err := p.Validate() + if tt.wantErr && err == nil { + t.Errorf("Validate() = nil, want an error for status %q", tt.status) + } + if !tt.wantErr && err != nil { + t.Errorf("Validate() = %v, want nil for status %q", err, tt.status) + } + }) + } +} diff --git a/internal/observatory/iteration_sink_test.go b/internal/observatory/iteration_sink_test.go new file mode 100644 index 000000000..49fd66cb1 --- /dev/null +++ b/internal/observatory/iteration_sink_test.go @@ -0,0 +1,324 @@ +package observatory + +import ( + "context" + "fmt" + "math" + "strings" + "testing" +) + +// M-MISSION-LOOP-UNIFIED-TELEMETRY M2/M3 — sink-level coverage of the iteration +// write path. +// +// The read-back tests in iteration_post_test.go go through real SQLite, which +// needs cgo. These exercise the SAME semantics through a fake sink so the +// accounting rules stay covered on any toolchain, and so the NODE-GENERIC claim +// is asserted rather than assumed: PostIterationTo must behave identically +// against a sink that is not the local SQLite store. + +// fakeSink models the subset of store semantics PostIterationTo depends on, +// mirroring store_chains.go: CreateStage defaults to pending, UpdateStageMetrics +// ACCUMULATES, and a stage transitioning to completed bumps the chain's +// stages_completed counter. +type fakeSink struct { + chains map[string]*ExecutionChain + stages map[string]*ChainStage + models map[string]string + // calls is the ordered method log, so ordering claims are asserted. + calls []string + // failOn makes the named method return an error (outage simulation). + failOn string + nextID int +} + +func newFakeSink() *fakeSink { + return &fakeSink{ + chains: map[string]*ExecutionChain{}, + stages: map[string]*ChainStage{}, + models: map[string]string{}, + } +} + +func (f *fakeSink) id(prefix string) string { + f.nextID++ + return fmt.Sprintf("%s-%d", prefix, f.nextID) +} + +func (f *fakeSink) record(method string) error { + f.calls = append(f.calls, method) + if f.failOn == method { + return fmt.Errorf("fakeSink: %s unavailable", method) + } + return nil +} + +func (f *fakeSink) CreateChain(_ context.Context, req *ChainCreateRequest) (*ExecutionChain, error) { + if err := f.record("CreateChain"); err != nil { + return nil, err + } + c := &ExecutionChain{ID: f.id("chain"), SourceType: req.SourceType, SourceRef: req.SourceRef} + f.chains[c.ID] = c + return c, nil +} + +func (f *fakeSink) CreateStage(_ context.Context, req *StageCreateRequest) (*ChainStage, error) { + if err := f.record("CreateStage"); err != nil { + return nil, err + } + s := &ChainStage{ + ID: f.id("stage"), + ChainID: req.ChainID, + AgentID: req.AgentID, + Provider: req.Provider, + Status: StageStatusPending, // the default this milestone exists to move off + } + f.stages[s.ID] = s + return s, nil +} + +func (f *fakeSink) UpdateStageMetrics(_ context.Context, stageID string, cost float64, tokensIn, tokensOut, turns, toolCalls int, durationMs int64, _ string) error { + if err := f.record("UpdateStageMetrics"); err != nil { + return err + } + s, ok := f.stages[stageID] + if !ok { + return fmt.Errorf("fakeSink: stage not found: %s", stageID) + } + s.Cost += cost + s.TokensIn += tokensIn + s.TokensOut += tokensOut + s.Turns += turns + s.ToolCalls += toolCalls + s.DurationMs += durationMs + return nil +} + +func (f *fakeSink) UpdateStageStatus(_ context.Context, stageID string, status ChainStageStatus) error { + if err := f.record("UpdateStageStatus"); err != nil { + return err + } + s, ok := f.stages[stageID] + if !ok { + return fmt.Errorf("fakeSink: stage not found: %s", stageID) + } + s.Status = status + if status == StageStatusCompleted { + if c, ok := f.chains[s.ChainID]; ok { + c.StagesCompleted++ + } + } + return nil +} + +func (f *fakeSink) UpdateChainMetrics(_ context.Context, id string, cost float64, tokens, turns int) error { + if err := f.record("UpdateChainMetrics"); err != nil { + return err + } + c, ok := f.chains[id] + if !ok { + return fmt.Errorf("fakeSink: chain not found: %s", id) + } + c.TotalCost += cost + c.TotalTokens += tokens + c.TotalTurns += turns + return nil +} + +// modelFakeSink additionally records stage models (what SQLite can do and +// Firestore cannot). +type modelFakeSink struct{ *fakeSink } + +func (f *modelFakeSink) UpdateStageEvalAssessment(_ context.Context, stageID string, a *EvalAssessment) error { + if err := f.record("UpdateStageEvalAssessment"); err != nil { + return err + } + f.models[stageID] = a.Model + return nil +} + +var ( + _ IterationSink = (*fakeSink)(nil) + _ IterationSink = (*modelFakeSink)(nil) + _ IterationModelSink = (*modelFakeSink)(nil) +) + +func (f *fakeSink) stageByAgentID(agentID string) *ChainStage { + for _, s := range f.stages { + if s.AgentID == agentID { + return s + } + } + return nil +} + +// iter190Post is the real measured shape: 4 stages, 3 providers, two stages +// carrying cost with ZERO tokens. +func iter190Post() *IterationPost { + return &IterationPost{ + Source: "manual:mission:v1/iter-190", + Stages: []IterationStage{ + {Role: "controller", Provider: "anthropic", QuotaBucket: "opus", Status: "completed"}, + {Role: "designer", Provider: "codex", QuotaBucket: "codex", Status: "completed"}, + {Role: "quorum-r1", Provider: "openrouter", Model: "gpt-5.6-sol", CostUSD: 0.0570, Status: "completed"}, + {Role: "quorum-r2", Provider: "openrouter", Model: "gemini-3.1-pro", CostUSD: 0.0507, Status: "completed"}, + }, + } +} + +func TestPostIterationTo_StatusIsPerStage(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + post := iter190Post() + post.Stages[1].Status = "failed" // the designer stage genuinely failed + + chainID, err := PostIterationTo(context.Background(), sink, post) + if err != nil { + t.Fatalf("PostIterationTo: %v", err) + } + + want := map[string]ChainStageStatus{ + "controller (quota:opus)": StageStatusCompleted, + "designer (quota:codex)": StageStatusFailed, + "quorum-r1": StageStatusCompleted, + "quorum-r2": StageStatusCompleted, + } + for agentID, wantStatus := range want { + st := sink.stageByAgentID(agentID) + if st == nil { + t.Fatalf("stage %q not written", agentID) + } + if st.Status != wantStatus { + t.Errorf("stage %q status = %q, want %q", agentID, st.Status, wantStatus) + } + } + + // stages_completed must exclude the failed stage — a blanket transition would + // read 4 here and hide the failure. + if got := sink.chains[chainID].StagesCompleted; got != 3 { + t.Errorf("StagesCompleted = %d, want 3 (the failed stage must not count)", got) + } +} + +func TestPostIterationTo_ChainTotalEqualsSumOfStages(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + post := iter190Post() + // One metered stage that DOES carry tokens, so the assertion covers both the + // cost-without-tokens rows and a normal one. + post.Stages[3].TokensIn = 9000 + post.Stages[3].TokensOut = 1000 + + chainID, err := PostIterationTo(context.Background(), sink, post) + if err != nil { + t.Fatalf("PostIterationTo: %v", err) + } + + var stageCost float64 + var stageTokens int + for _, s := range sink.stages { + stageCost += s.Cost + stageTokens += s.TokensIn + s.TokensOut + } + chain := sink.chains[chainID] + if math.Abs(chain.TotalCost-stageCost) > 1e-9 { + t.Errorf("chain TotalCost = %v, want %v (sum of stages)", chain.TotalCost, stageCost) + } + if chain.TotalTokens != stageTokens { + t.Errorf("chain TotalTokens = %d, want %d (sum of stages)", chain.TotalTokens, stageTokens) + } + // The measured regression: iter-190 held $0.1077 and reported $0.0000. + if math.Abs(chain.TotalCost-0.1077) > 1e-9 { + t.Errorf("chain TotalCost = %v, want 0.1077 (the iter-190 figure)", chain.TotalCost) + } +} + +func TestPostIterationTo_OmittedStatusStaysPending(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + post := iter190Post() + for i := range post.Stages { + post.Stages[i].Status = "" // an unversioned payload from an older skill + } + + chainID, err := PostIterationTo(context.Background(), sink, post) + if err != nil { + t.Fatalf("PostIterationTo on an unversioned payload: %v", err) + } + for _, s := range sink.stages { + if s.Status != StageStatusPending { + t.Errorf("stage %q status = %q, want pending (no status supplied)", s.AgentID, s.Status) + } + } + for _, c := range sink.calls { + if c == "UpdateStageStatus" { + t.Error("UpdateStageStatus called for a payload that supplied no status") + } + } + // Aggregation is independent of status and still happens. + if math.Abs(sink.chains[chainID].TotalCost-0.1077) > 1e-9 { + t.Errorf("TotalCost = %v, want 0.1077", sink.chains[chainID].TotalCost) + } +} + +// TestPostIterationTo_StatusWrittenAfterMetrics pins the ordering: the status +// transition stamps completed_at, so it must land after the stage is credited. +func TestPostIterationTo_StatusWrittenAfterMetrics(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + if _, err := PostIterationTo(context.Background(), sink, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{{Role: "quorum-r1", Provider: "openrouter", Model: "gpt-5.6-sol", CostUSD: 0.05, TokensIn: 10, TokensOut: 5, Status: "completed"}}, + }); err != nil { + t.Fatalf("PostIterationTo: %v", err) + } + + got := strings.Join(sink.calls, ",") + want := "CreateChain,CreateStage,UpdateStageMetrics,UpdateStageEvalAssessment,UpdateStageStatus,UpdateChainMetrics" + if got != want { + t.Errorf("call order = %q, want %q", got, want) + } +} + +// TestPostIterationTo_SinkWithoutModelSupport is the node-generic case: the +// Firestore observatory does not implement eval_assessment. Dropping the model +// must not fail the post (the cloud leg would never succeed) and must not be +// silent (PostIterationTo reports it on stderr). +func TestPostIterationTo_SinkWithoutModelSupport(t *testing.T) { + sink := newFakeSink() // no UpdateStageEvalAssessment + chainID, err := PostIterationTo(context.Background(), sink, iter190Post()) + if err != nil { + t.Fatalf("PostIterationTo against a model-less sink: %v", err) + } + if math.Abs(sink.chains[chainID].TotalCost-0.1077) > 1e-9 { + t.Errorf("TotalCost = %v, want 0.1077 — accounting must not depend on model support", sink.chains[chainID].TotalCost) + } + for _, c := range sink.calls { + if c == "UpdateStageEvalAssessment" { + t.Error("UpdateStageEvalAssessment called on a sink that does not implement it") + } + } +} + +// TestPostIterationTo_WriteFailureIsReported keeps the spool contract honest: a +// mid-post failure must surface as an error so the caller buffers the WHOLE post. +func TestPostIterationTo_WriteFailureIsReported(t *testing.T) { + for _, method := range []string{"CreateChain", "CreateStage", "UpdateStageMetrics", "UpdateStageStatus", "UpdateChainMetrics"} { + t.Run(method, func(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + sink.failOn = method + if _, err := PostIterationTo(context.Background(), sink, iter190Post()); err == nil { + t.Errorf("PostIterationTo = nil error, want a failure when %s is unavailable", method) + } + }) + } +} + +func TestPostIterationTo_ValidationRejectsBadPost(t *testing.T) { + sink := &modelFakeSink{newFakeSink()} + if _, err := PostIterationTo(context.Background(), sink, &IterationPost{ + Source: "mission:v1/iter-191", + Stages: []IterationStage{{Role: "controller", Status: "finished"}}, + }); err == nil { + t.Error("PostIterationTo accepted an unknown stage status; want a loud rejection") + } + if len(sink.calls) != 0 { + t.Errorf("validation ran after writes began: %v", sink.calls) + } +} diff --git a/internal/storage/firestore/observatory.go b/internal/storage/firestore/observatory.go index 617b13537..22862c33e 100644 --- a/internal/storage/firestore/observatory.go +++ b/internal/storage/firestore/observatory.go @@ -15,6 +15,14 @@ import ( // Compile-time check that ObservatoryStore implements observatory.Backend. var _ obs.Backend = (*ObservatoryStore)(nil) +// This store is the CLOUD leg of the mission-iteration dual-write +// (M-MISSION-LOOP-UNIFIED-TELEMETRY M3). IterationSink is deliberately a subset +// of Backend so that stays true; this assertion is what fails if it stops being +// one. It does NOT implement IterationModelSink — eval_assessment is a +// SQLite-only write — which is why PostIterationTo treats that as optional and +// reports the drop rather than requiring it. +var _ obs.IterationSink = (*ObservatoryStore)(nil) + // Observatory Firestore collection names (prefixed with obs_ to avoid collisions). const ( collObsWorkspaces = "obs_workspaces" From 4d29d56daa148cb2ea1d88084f412ce30a5e5e02 Mon Sep 17 00:00:00 2001 From: ailang-agent Date: Thu, 13 Aug 2026 13:13:30 +0000 Subject: [PATCH 2/2] refactor(observatory): plain if/else for the optional model sink Matches the surrounding style in iteration_post.go; no behaviour change. Co-Authored-By: Claude Opus 5 --- internal/observatory/iteration_post.go | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/internal/observatory/iteration_post.go b/internal/observatory/iteration_post.go index 0972ef847..3bfebc533 100644 --- a/internal/observatory/iteration_post.go +++ b/internal/observatory/iteration_post.go @@ -188,16 +188,13 @@ func PostIterationTo(ctx context.Context, store IterationSink, p *IterationPost) } // Record the model (metered lanes) so the M1 classifier can resolve a rate. if st.Model != "" { - switch { - case canRecordModel: - if err := modelSink.UpdateStageEvalAssessment(ctx, stage.ID, &EvalAssessment{ - Model: st.Model, - EvalMode: "mission", - }); err != nil { - return chain.ID, fmt.Errorf("update stage %d model: %w", i, err) - } - default: + if !canRecordModel { modelsDropped++ + } else if err := modelSink.UpdateStageEvalAssessment(ctx, stage.ID, &EvalAssessment{ + Model: st.Model, + EvalMode: "mission", + }); err != nil { + return chain.ID, fmt.Errorf("update stage %d model: %w", i, err) } } // Status LAST, so completed_at lands after the stage is fully credited.