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 <<JSON | ailang chains post-iteration || true   # `|| true`: telemetry NEVER blocks the loop
 {
   "source": "mission:${MISSION_NAME:-v1}/iter-${ITER}",
   "stages": [
-    {"role":"executor","provider":"codex","model":"<model>","cost_usd":<metered $>,"tokens_in":<n>,"tokens_out":<n>},
-    {"role":"controller","quota_bucket":"opus"},
-    {"role":"evaluator","quota_bucket":"sonnet"}
+    {"role":"executor","provider":"codex","model":"<model>","cost_usd":<metered $>,"tokens_in":<n>,"tokens_out":<n>,"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 `<role>
-(quota:<bucket>)` — 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:<bucket>)` — 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 <chain-id>       # 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 <id> --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 <chain-id>   # Quick issue check")
 		fmt.Println("  ailang chains diff <chain-id>        # 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 <task-id>     Show streaming execution logs\n", cyan("coordinator logs"))
 	fmt.Printf("  %s <cmd>              View execution chains (task→session→chat linkage)\n", cyan("chains"))
 	fmt.Printf("    %s                  List all chains\n", cyan("chains list"))
-	fmt.Printf("    %s <chain-id>       View chain stages and details\n", cyan("chains view"))
+	fmt.Printf("    %s <chain-id>       View chain stages and details (--remote reads cloud)\n", cyan("chains view"))
 	fmt.Printf("    %s <chain-id>       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 <chain-id>  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 <cmd>            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 <iter-191> --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..3bfebc533 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 "<role> (quota:<bucket>)" 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,39 @@ 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{
+			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.
+		// 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"