rust mode: keep the pristine module array as the delta basis; hold the demoted signed assistant's native vector until a priced pass (ASTRO 05:32Z busts)

Co-authored-by: Alfonso <alfonso-magic-context@users.noreply.github.com>

Author: ualtinok

crates/mc-module/src/lib.rs

@@ -12202,7 +12202,7 @@ fn attach_native_messages_with_tags(
         return;
     }
     let native_messages = encode_full_native_messages(
-        response.messages(),
+        response,
         request,
         reasoning_watermark,
         tag_numbers,
@@ -12388,14 +12388,16 @@ fn native_reasoning_should_clear(
 }
 
 fn encode_full_native_messages(
-    served: &[transform::ServedMessage],
+    response: &transform::TransformResponse,
     request: &TransformRequest,
     reasoning_watermark: u64,
     tag_numbers: &BTreeMap<String, u64>,
     mutation_exempt_mid: Option<&str>,
     lineage_anchor_mid: Option<&str>,
     transition_consumed: bool,
 ) -> Vec<Value> {
+    let served = response.messages();
+    let native_reasoning_keep_mids = &response.native_reasoning_keep_mids;
     let sidecar = request
         .native_messages
         .as_deref()
@@ -12433,9 +12435,30 @@ fn encode_full_native_messages(
             tag_numbers,
         );
     }
+    replay_native_reasoning_keeps(&mut native_messages, &sidecar, native_reasoning_keep_mids);
     native_messages
 }
 
+fn replay_native_reasoning_keeps(
+    messages: &mut [Value],
+    sidecar: &codec::DecodeSidecar,
+    keep_mids: &[String],
+) {
+    for message in messages {
+        let Some(mid) = message["info"]["id"].as_str() else {
+            continue;
+        };
+        if !keep_mids.iter().any(|keep| keep == mid) {
+            continue;
+        }
+        if let Some(meta) = sidecar.messages.get(mid) {
+            if let Some(parts) = meta.raw.get("parts") {
+                message["parts"] = parts.clone();
+            }
+        }
+    }
+}
+
 fn native_ingress_chunks(
     request: &TransformRequest,
     encoded_chunks: &[NativeEncodedChunk],
@@ -12599,7 +12622,13 @@ fn attach_native_messages_incremental(
             }
             Some(hash)
         });
-        let mutation_exempt = slot.is_some_and(|mid| mutation_exempt_mids.contains(&mid));
+        let mutation_exempt = slot.is_some_and(|mid| {
+            mutation_exempt_mids.contains(&mid)
+                || response
+                    .native_reasoning_keep_mids
+                    .iter()
+                    .any(|keep| keep == mid)
+        });
         let reasoning_exempt = slot.is_some_and(|mid| Some(mid) == newest_assistant_mid);
         let (tag_number, reasoning_should_clear) = native_reasoning_should_clear(
             served,
@@ -12713,6 +12742,11 @@ fn attach_native_messages_incremental(
             tag_numbers,
         );
     }
+    replay_native_reasoning_keeps(
+        &mut suffix_values,
+        &sidecar,
+        &response.native_reasoning_keep_mids,
+    );
     chunks.extend(
         encoded_suffix
             .into_iter()
@@ -12734,7 +12768,7 @@ fn attach_native_messages_incremental(
 
     if native_attachment_differential_enabled() {
         let full = encode_full_native_messages(
-            response.messages(),
+            response,
             request,
             reasoning_watermark,
             tag_numbers,
@@ -12845,7 +12879,7 @@ fn finalize_native_messages_response(
         fallback_reason.get_or_insert(NativeDeltaFallbackReason::MissingNativeContent);
         response.native_messages = Some(
             encode_full_native_messages(
-                response.messages(),
+                response,
                 request,
                 reasoning_watermark,
                 tag_numbers,
@@ -21708,6 +21742,103 @@ mod tests {
             .all(|block| block.synthetic));
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn post_hard_defer_preserves_demoted_native_thinking_until_priced_pass() {
+        let producer = Arc::new(ProducerState::default());
+        let (handler, store, _dir, _project) = handler_with_store(producer, default_test_config());
+        let mut native = vec![
+            json!({"info":{"id":"user","role":"user"},"parts":[{"type":"text","text":"start"}]}),
+            json!({"info":{"id":"previous","role":"assistant"},"parts":[{"type":"text","text":"earlier step"}]}),
+        ];
+        let make_request = |native: &Vec<Value>, model: &str| {
+            let decoded = codec::decode_opencode(native);
+            let mut req = request(decoded.messages);
+            req["serializer_profile"] = json!("opencode-aisdk");
+            req["serve_native"] = json!(true);
+            req["tool_present"] = json!(true);
+            req["provider_id"] = json!("anthropic");
+            req["model_key"] = json!(model);
+            req["mid_turn"] = json!(false);
+            req["native_messages"] = json!(native);
+            req
+        };
+        let warm = call_transform_request(&handler, make_request(&native, "claude-opus-5")).await;
+        assert_eq!(warm["action"], "HARD");
+        let thinking = json!({"type":"reasoning","text":"signed thinking", "metadata":{"anthropic":{"signature":"fixture-signature"}}});
+        native.push(json!({"info":{"id":"thinking","role":"assistant"},"parts":[
+            thinking.clone(), {"type":"text","text":"dispatching"},
+            {"type":"tool","tool":"work","callID":"call-thinking","state":{"status":"completed","input":{"action":"run"},"output":"dispatched"}}
+        ]}));
+        native.push(json!({"info":{"id":"continue","role":"user"},"parts":[{"type":"text","text":"continue"}]}));
+        let hard =
+            call_transform_request(&handler, make_request(&native, "claude-fable-5-1")).await;
+        assert_eq!(hard["action"], "HARD");
+        let mut previous = hard["native_messages"].as_array().unwrap().clone();
+        let hash = |messages: &[Value]| -> String {
+            format!(
+                "{:x}",
+                Sha256::digest(serde_json::to_vec(messages).unwrap())
+            )
+        };
+        for step in 0..2 {
+            native.push(json!({"info":{"id":format!("step-{step}"),"role":"assistant"},"parts":[
+                {"type":"reasoning","text":format!("adaptive step {step}"),"metadata":{"anthropic":{"signature":format!("signature-{step}")}}},
+                {"type":"tool","tool":"work","callID":format!("call-{step}"),"state":{"status":"completed","input":{"action":"show"},"output":"done"}}
+            ]}));
+            let deferred =
+                call_transform_request(&handler, make_request(&native, "claude-fable-5-1")).await;
+            assert_eq!(deferred["action"], "SOFT+");
+            let served = deferred["native_messages"].as_array().unwrap();
+            assert!(served.len() > previous.len());
+            assert_eq!(
+                hash(&served[..previous.len()]),
+                hash(&previous),
+                "demotion must not rewrite the served prefix: {:?}",
+                served
+                    .iter()
+                    .zip(&previous)
+                    .enumerate()
+                    .filter(|(_, (new, old))| new != old)
+                    .collect::<Vec<_>>()
+            );
+            let target = served
+                .iter()
+                .find(|message| message["info"]["id"] == "thinking")
+                .unwrap();
+            assert_eq!(target["parts"][0], thinking);
+            assert!(!store
+                .load("ses")
+                .unwrap()
+                .core
+                .frozen_units
+                .iter()
+                .any(|unit| unit.key == "strip:merged_reasoning:thinking"));
+            previous = served.clone();
+        }
+        let mut priced_request = make_request(&native, "claude-fable-5-1");
+        priced_request["render_config"] = json!("independent-priced-config-change");
+        let priced = call_transform_request(&handler, priced_request).await;
+        assert_eq!(priced["action"], "HARD");
+        let target = priced["native_messages"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|message| message["info"]["id"] == "thinking")
+            .unwrap();
+        assert!(target["parts"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .all(|part| part["type"] != "reasoning"));
+        assert!(store
+            .load("ses")
+            .unwrap()
+            .core
+            .frozen_units
+            .iter()
+            .any(|unit| unit.key == "strip:merged_reasoning:thinking"));
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn native_attachment_reuses_transform_tag_baseline_and_preserves_bytes() {
         let producer = Arc::new(ProducerState::default());

crates/mc-module/src/transform.rs

@@ -1473,6 +1473,9 @@ pub struct NativeMessagesDelta {
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct TransformResponse {
+    /// Native vectors protected while newest remain unchanged until the next priced pass.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub native_reasoning_keep_mids: Vec<String>,
     pub status: TransformStatus,
     pub served_from: ServedFrom,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -1569,6 +1572,7 @@ impl TransformResponse {
 
     pub fn need_full_sync(full_array_fingerprint: Option<String>) -> Self {
         Self {
+            native_reasoning_keep_mids: Vec::new(),
             status: TransformStatus::NeedFullSync,
             served_from: ServedFrom::Transform,
             full_array_fingerprint,
@@ -1606,6 +1610,7 @@ impl TransformResponse {
         full_array_fingerprint: Option<String>,
     ) -> Self {
         Self {
+            native_reasoning_keep_mids: Vec::new(),
             status: TransformStatus::Ok,
             served_from: ServedFrom::Transform,
             full_array_fingerprint,
@@ -3094,6 +3099,15 @@ fn apply_additive_only(
         mutation_exempt_mid: None,
         lineage_anchor_mid: None,
         response: TransformResponse {
+            native_reasoning_keep_mids: core
+                .frozen_units
+                .iter()
+                .filter_map(|unit| {
+                    unit.key
+                        .strip_prefix("strip:native_reasoning_keep:")
+                        .map(str::to_owned)
+                })
+                .collect(),
             status: TransformStatus::Ok,
             served_from: ServedFrom::Transform,
             full_array_fingerprint: req.full_array_fingerprint.clone(),
@@ -5347,6 +5361,26 @@ fn apply_once(
         no_trim_meta = Some(output_meta);
     }
     let output_meta = no_trim_meta.as_ref().unwrap_or(&meta);
+    // OpenCode must replay the newest signed assistant's complete native vector. Its
+    // demotion is not permission to apply previously withheld overlays on a defer.
+    // Release these keeps only when the prefix is already being repriced.
+    if is_provider_prefix_mutation_pass {
+        core.frozen_units
+            .retain(|unit| !unit.key.starts_with("strip:native_reasoning_keep:"));
+    }
+    if serializer_profile == Some(SerializerProfile::OpencodeAiSdk) && req.serve_native {
+        if let Some(mid) = latest_assistant_reasoning_mutation_exempt_mid(&req.messages) {
+            if req.messages.iter().any(|message| {
+                message.mid == mid && message.ck.content.iter().any(is_reasoning_block)
+            }) {
+                let key = format!("strip:native_reasoning_keep:{mid}");
+                if !core.frozen_units.iter().any(|unit| unit.key == key) {
+                    core.frozen_units
+                        .push(strip_unit("native_reasoning_keep", mid, ""));
+                }
+            }
+        }
+    }
     let output_cache_snapshot = output_cache.map(|cache| {
         cache
             .lock()
@@ -5676,6 +5710,15 @@ fn apply_once(
         mutation_exempt_mid: mutation_exempt_mid.map(str::to_string),
         lineage_anchor_mid: lineage_anchor_mid.map(str::to_string),
         response: TransformResponse {
+            native_reasoning_keep_mids: core
+                .frozen_units
+                .iter()
+                .filter_map(|unit| {
+                    unit.key
+                        .strip_prefix("strip:native_reasoning_keep:")
+                        .map(str::to_owned)
+                })
+                .collect(),
             status: TransformStatus::Ok,
             served_from: ServedFrom::Transform,
             full_array_fingerprint: req.full_array_fingerprint.clone(),

packages/plugin/scripts/analyze-cache-busts.test.ts

@@ -102,6 +102,27 @@ function snapshotsFor(dir: string, session: string) {
 }
 
 describe("analyze-cache-bust dump discovery", () => {
+    test("prints complete UTF-8 body bytes separately from reusable normalized prefix bytes", () => {
+        const dir = mkdtempSync(join(tmpdir(), "cache-bust-body-bytes-"));
+        tempDirs.push(dir);
+        const session = "ses_bodyBytes";
+        const previous = bodyWithTailCheckpoint("old § tail");
+        const current = bodyWithTailCheckpoint("new § tail with more bytes");
+        const previousStem = `2026-09-02T08-41-00-000Z-${session}`;
+        const currentStem = `2026-09-02T08-42-00-000Z-${session}`;
+        writeDump(dir, previousStem, "2026-09-02T08:41:00Z", session, previous);
+        writeDump(dir, currentStem, "2026-09-02T08:42:00Z", session, current);
+        const prettyBody = `${JSON.stringify(current, null, 2)}\n`;
+        writeFileSync(join(dir, `${currentStem}.body.json`), prettyBody);
+        const expected = [Buffer.byteLength(JSON.stringify(previous)), Buffer.byteLength(prettyBody)];
+        expect(snapshotsFor(dir, session).map((snapshot) => snapshot.bodyBytes)).toEqual(expected);
+        const run = Bun.spawnSync([process.execPath, join(import.meta.dir, "analyze-cache-busts.ts"), "--session", session, "--dir", dir, "--all-rows"]);
+        expect(run.exitCode).toBe(0);
+        const output = run.stdout.toString();
+        expect(output).toContain("prevBodyBytes → curBodyBytes");
+        expect(output).toContain("reusableNormalizedPrefix@breakpoint");
+        expect(output).toContain(`${expected[0]!.toLocaleString()}B → ${expected[1]!.toLocaleString()}B`);
+    });
     test("loads and orders legacy and sequence+routing filename layouts by timestamp and sequence", () => {
         const dir = mkdtempSync(join(tmpdir(), "cache-bust-fixture-"));
         tempDirs.push(dir);

packages/plugin/scripts/analyze-cache-busts.ts

@@ -53,6 +53,7 @@ interface MeterUsage {
 interface Snapshot {
     file: string;
     bodyPath: string;
+    bodyBytes: number;
     createdAt: string;
     session: string;
     messagesCount: number;
@@ -350,8 +351,11 @@ function loadSnapshots(opts: ReturnType<typeof parseArgs>): Snapshot[] {
         if (!bodyPath) continue;
 
         let body: Json;
+        let bodyBytes: number;
         try {
-            body = JSON.parse(readFileSync(bodyPath, "utf8")) as Json;
+            const rawBody = readFileSync(bodyPath);
+            bodyBytes = rawBody.byteLength;
+            body = JSON.parse(rawBody.toString("utf8")) as Json;
         } catch {
             continue;
         }
@@ -367,6 +371,7 @@ function loadSnapshots(opts: ReturnType<typeof parseArgs>): Snapshot[] {
         snaps.push({
             file: metaFile,
             bodyPath,
+            bodyBytes,
             createdAt,
             session,
             messagesCount: typeof bodyMeta?.messagesCount === "number" ? bodyMeta.messagesCount : -1,
@@ -554,7 +559,7 @@ function main(): void {
     console.log("Dashboard times are local (UTC+2); table times are UTC.");
     console.log("Meter rule: shortRead when cacheRead + current input < prevTotal - ε, where ε=max(64, previous input). A short read is BUST only when bytes diverge before the current tail breakpoint; otherwise it is LATENCY.");
     console.log(
-        "time(UTC)          | segs | verdict          | meter                                                  | meterVsBytes | first-divergence                | prevBytes → curBytes        | cachedPrefix@breakpoint",
+        "time(UTC)          | segs | verdict          | meter                                                  | meterVsBytes | first-divergence                | prevBodyBytes → curBodyBytes | reusableNormalizedPrefix@breakpoint",
     );
     console.log(
         "-------------------|------|------------------|--------------------------------------------------------|--------------|---------------------------------|-----------------------------|------------------------",
@@ -586,9 +591,8 @@ function main(): void {
         const attribution = segment
             ? `${segment.id} (bytes ${row.byteVerdict})`
             : `(identical; bytes ${row.byteVerdict})`;
-        const previousPrefix = cachedPrefixBytes(previous.segments, previous.segments.length);
         const currentPrefix = cachedPrefixBytes(row.current.segments, index);
-        const byteDelta = `${previousPrefix.bytes.toLocaleString()}B → ${currentPrefix.bytes.toLocaleString()}B`;
+        const byteDelta = `${previous.bodyBytes.toLocaleString()}B → ${row.current.bodyBytes.toLocaleString()}B`;
         const verdictLabel =
             row.verdict === "UNMETERED"
                 ? `UNMETERED (bytes ${row.byteVerdict})`