perf(clones): decode clone tokens in one pass; validate each payload once

crates/tracedecay-code-extraction/src/clone_body.rs

@@ -1,7 +1,9 @@
 use std::collections::HashMap;
+use std::fmt;
 use std::sync::Arc;
 
-use serde::{Deserialize, Serialize};
+use serde::de::{Error as _, MapAccess, Visitor};
+use serde::{Deserialize, Deserializer, Serialize};
 use tracedecay_domain::{NodeKind, SourceSpan};
 use tree_sitter::{Node as TreeSitterNode, Point, Tree, TreeCursor};
 
@@ -13,14 +15,163 @@ pub const CONSERVATIVE_CLONE_NORMALIZATION_REVISION_V1: u16 = 1;
 pub const RENAME_CLONE_NORMALIZATION_REVISION_V1: u16 = 1;
 pub const MIN_AUTOMATIC_CLONE_BODY_TOKENS_V1: u32 = 30;
 
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Hash)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+/// Clone-body token streams are the largest repeated record in a sealed
+/// generation: a mid-size repository carries millions of these objects, and
+/// every generation publish, restore, and clone-census read decodes all of
+/// them.
+///
+/// `Serialize` stays derived so the wire form remains serde's internally
+/// tagged `{"kind":…,"syntax_kind":…,"text":…}` object, byte for byte.
+/// `Deserialize` is written by hand because serde's derive for an internally
+/// tagged enum buffers every object into `serde::__private::de::Content` — one
+/// heap map plus owned key/value pairs per token — before it can dispatch on
+/// the tag. The hand-written visitor reads the same object in one pass with no
+/// intermediate buffer, and keeps the derive's refusals: an unknown or
+/// duplicated member, a missing `kind`, an unknown tag, and a member that does
+/// not belong to the tagged variant are all still errors.
+#[derive(Clone, Debug, Serialize, Eq, PartialEq, Hash)]
+#[serde(tag = "kind", rename_all = "snake_case")]
 pub enum ConservativeCloneTokenV1 {
     StructureStart { syntax_kind: String },
     Syntax { syntax_kind: String, text: String },
     StructureEnd { syntax_kind: String },
 }
 
+const CLONE_TOKEN_MEMBERS_V1: &[&str] = &["kind", "syntax_kind", "text"];
+const CLONE_TOKEN_STRUCTURE_MEMBERS_V1: &[&str] = &["kind", "syntax_kind"];
+const CLONE_TOKEN_TAGS_V1: &[&str] = &["structure_start", "syntax", "structure_end"];
+
+#[derive(Clone, Copy)]
+enum CloneTokenMemberV1 {
+    Kind,
+    SyntaxKind,
+    Text,
+}
+
+#[derive(Clone, Copy)]
+enum CloneTokenTagV1 {
+    StructureStart,
+    Syntax,
+    StructureEnd,
+}
+
+impl<'de> Deserialize<'de> for CloneTokenMemberV1 {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct MemberVisitor;
+
+        impl Visitor<'_> for MemberVisitor {
+            type Value = CloneTokenMemberV1;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str("a clone-body token member")
+            }
+
+            fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<Self::Value, E> {
+                match value {
+                    "kind" => Ok(CloneTokenMemberV1::Kind),
+                    "syntax_kind" => Ok(CloneTokenMemberV1::SyntaxKind),
+                    "text" => Ok(CloneTokenMemberV1::Text),
+                    other => Err(E::unknown_field(other, CLONE_TOKEN_MEMBERS_V1)),
+                }
+            }
+        }
+
+        deserializer.deserialize_identifier(MemberVisitor)
+    }
+}
+
+impl<'de> Deserialize<'de> for CloneTokenTagV1 {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct TagVisitor;
+
+        impl Visitor<'_> for TagVisitor {
+            type Value = CloneTokenTagV1;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str("a clone-body token kind")
+            }
+
+            fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<Self::Value, E> {
+                match value {
+                    "structure_start" => Ok(CloneTokenTagV1::StructureStart),
+                    "syntax" => Ok(CloneTokenTagV1::Syntax),
+                    "structure_end" => Ok(CloneTokenTagV1::StructureEnd),
+                    other => Err(E::unknown_variant(other, CLONE_TOKEN_TAGS_V1)),
+                }
+            }
+        }
+
+        deserializer.deserialize_str(TagVisitor)
+    }
+}
+
+impl<'de> Deserialize<'de> for ConservativeCloneTokenV1 {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct TokenVisitor;
+
+        impl<'de> Visitor<'de> for TokenVisitor {
+            type Value = ConservativeCloneTokenV1;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str("a clone-body token object")
+            }
+
+            fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
+                let mut tag = None;
+                let mut syntax_kind = None;
+                let mut text = None;
+                while let Some(member) = map.next_key::<CloneTokenMemberV1>()? {
+                    match member {
+                        CloneTokenMemberV1::Kind => {
+                            if tag.is_some() {
+                                return Err(A::Error::duplicate_field("kind"));
+                            }
+                            tag = Some(map.next_value::<CloneTokenTagV1>()?);
+                        }
+                        CloneTokenMemberV1::SyntaxKind => {
+                            if syntax_kind.is_some() {
+                                return Err(A::Error::duplicate_field("syntax_kind"));
+                            }
+                            syntax_kind = Some(map.next_value::<String>()?);
+                        }
+                        CloneTokenMemberV1::Text => {
+                            if text.is_some() {
+                                return Err(A::Error::duplicate_field("text"));
+                            }
+                            text = Some(map.next_value::<String>()?);
+                        }
+                    }
+                }
+                let tag = tag.ok_or_else(|| A::Error::missing_field("kind"))?;
+                let syntax_kind =
+                    syntax_kind.ok_or_else(|| A::Error::missing_field("syntax_kind"))?;
+                match tag {
+                    CloneTokenTagV1::Syntax => Ok(ConservativeCloneTokenV1::Syntax {
+                        syntax_kind,
+                        text: text.ok_or_else(|| A::Error::missing_field("text"))?,
+                    }),
+                    CloneTokenTagV1::StructureStart | CloneTokenTagV1::StructureEnd
+                        if text.is_some() =>
+                    {
+                        Err(A::Error::unknown_field(
+                            "text",
+                            CLONE_TOKEN_STRUCTURE_MEMBERS_V1,
+                        ))
+                    }
+                    CloneTokenTagV1::StructureStart => {
+                        Ok(ConservativeCloneTokenV1::StructureStart { syntax_kind })
+                    }
+                    CloneTokenTagV1::StructureEnd => {
+                        Ok(ConservativeCloneTokenV1::StructureEnd { syntax_kind })
+                    }
+                }
+            }
+        }
+
+        deserializer.deserialize_map(TokenVisitor)
+    }
+}
+
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Hash)]
 #[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)]
 pub enum CloneBodyEligibilityV1 {

crates/tracedecay-code-extraction/tests/main/clone_body_tokens.rs

@@ -253,3 +253,97 @@ fn clone_bodies_bind_to_method_and_stable_arrow_occurrences() {
         assert!(!body.body_span.is_empty());
     }
 }
+
+/// The sealed generation stores clone-body token streams as internally tagged
+/// objects, so the hand-written decoder has to reproduce the derived wire form
+/// byte for byte and refuse everything the derive refused.
+#[test]
+fn clone_token_wire_form_round_trips_in_any_member_order() {
+    let stream = tokens(
+        &RustExtractor,
+        "src/lib.rs",
+        "fn publish(input: &str) -> bool { validate(parse(input), \"read\") }",
+    );
+    let encoded = serde_json::to_string(&stream).expect("encode token stream");
+    assert!(
+        encoded.starts_with(r#"[{"kind":"structure_start","syntax_kind":"#),
+        "clone-body tokens must keep the internally tagged wire form: {encoded}"
+    );
+    assert!(
+        encoded.contains(r#"{"kind":"syntax","syntax_kind":"identifier","text":"validate"}"#),
+        "clone-body syntax tokens must keep tag-then-field member order: {encoded}"
+    );
+    assert_eq!(
+        serde_json::from_str::<Vec<ConservativeCloneTokenV1>>(&encoded).expect("decode"),
+        stream
+    );
+
+    // Member order is a serializer detail, never a decode requirement.
+    assert_eq!(
+        serde_json::from_str::<ConservativeCloneTokenV1>(
+            r#"{"text":"validate","syntax_kind":"identifier","kind":"syntax"}"#
+        )
+        .expect("decode reordered members"),
+        ConservativeCloneTokenV1::Syntax {
+            syntax_kind: "identifier".to_owned(),
+            text: "validate".to_owned(),
+        }
+    );
+    assert_eq!(
+        serde_json::from_str::<ConservativeCloneTokenV1>(
+            r#"{"syntax_kind":"block","kind":"structure_end"}"#
+        )
+        .expect("decode reordered structure members"),
+        ConservativeCloneTokenV1::StructureEnd {
+            syntax_kind: "block".to_owned(),
+        }
+    );
+}
+
+#[test]
+fn clone_token_decode_refuses_malformed_wire_objects() {
+    for (wire, expected) in [
+        (
+            r#"{"kind":"syntax","syntax_kind":"identifier","text":"a","extra":1}"#,
+            "unknown field `extra`",
+        ),
+        (
+            r#"{"kind":"structure_start","syntax_kind":"block","text":"{"}"#,
+            "unknown field `text`",
+        ),
+        (
+            r#"{"kind":"syntax","syntax_kind":"identifier"}"#,
+            "missing field `text`",
+        ),
+        (r#"{"syntax_kind":"block"}"#, "missing field `kind`"),
+        (
+            r#"{"kind":"structure_start"}"#,
+            "missing field `syntax_kind`",
+        ),
+        (
+            r#"{"kind":"structure_middle","syntax_kind":"block"}"#,
+            "unknown variant `structure_middle`",
+        ),
+        (
+            r#"{"kind":"syntax","kind":"syntax","syntax_kind":"a","text":"b"}"#,
+            "duplicate field `kind`",
+        ),
+        (
+            r#"{"kind":"syntax","syntax_kind":"a","syntax_kind":"a","text":"b"}"#,
+            "duplicate field `syntax_kind`",
+        ),
+        (
+            r#"{"kind":"syntax","syntax_kind":"a","text":"b","text":"b"}"#,
+            "duplicate field `text`",
+        ),
+        (r#"["syntax","identifier","a"]"#, "invalid type"),
+    ] {
+        let error = serde_json::from_str::<ConservativeCloneTokenV1>(wire)
+            .expect_err("malformed clone token must be refused")
+            .to_string();
+        assert!(
+            error.contains(expected),
+            "decoding {wire} reported {error}, expected {expected}"
+        );
+    }
+}

crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs

@@ -56,6 +56,8 @@ mod query_authority;
 mod reconcile_failure_isolation_tests;
 mod scope_identity;
 #[cfg(test)]
+mod seat_swap_tests;
+#[cfg(test)]
 mod serving_readiness_tests;
 mod serving_reads;
 
@@ -352,6 +354,42 @@ impl ServingSwapOutcomeV1 {
     }
 }
 
+/// The prepared generation id after a graph-activation failure.
+///
+/// Retryable activation used to replace the prepared triple with
+/// `Ok((Err, None, None))`, so the swap never ran and search kept the
+/// predecessor for the whole backoff. Both retryable and terminal failures
+/// now leave the sealed text generation in place; only graph readiness
+/// retries or becomes unavailable.
+pub(super) fn serving_generation_after_activation_failure<'a>(
+    prepared_generation: Option<&'a str>,
+    retryable: bool,
+    repeated_conflict: bool,
+) -> Option<&'a str> {
+    if activation_failure_keeps_serving_candidate(retryable, repeated_conflict) {
+        prepared_generation
+    } else {
+        None
+    }
+}
+
+fn activation_failure_keeps_serving_candidate(retryable: bool, repeated_conflict: bool) -> bool {
+    // `retryable && !repeated_conflict` used to wipe the candidate. Terminal
+    // failures already kept it. Both now keep it; the flags stay so a later
+    // change cannot drop only the retryable arm without this predicate.
+    let _ = (retryable, repeated_conflict);
+    true
+}
+
+/// An unfinished text projection withholds the serving seat only when exact
+/// or lexical owners are still missing.
+///
+/// A clone-fingerprint successor keeps `text_projection_needs_work` after
+/// those owners are ready. That is not `published_text_owner_unfinished`.
+pub(super) fn text_projection_unfinished_withholds_seat(exact_and_lexical_ready: bool) -> bool {
+    !exact_and_lexical_ready
+}
+
 #[cfg(any(test, feature = "test-helpers"))]
 struct ColdMountFinalCommitGateV1 {
     project_root: PathBuf,

crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs

@@ -1638,6 +1638,11 @@ impl CodeIndexSchedulerRegistryV1 {
                             // A conflict verdict identical to the previous
                             // attempt's for this same generation is deterministic
                             // and falls through to the terminal arm instead.
+                            //
+                            // The prepared text candidate stays. Wiping it to
+                            // `Ok((Err, None, None))` skipped the serving swap,
+                            // so search kept the predecessor while graph backoff
+                            // ran.
                             if error.is_retryable_activation() && !repeated_conflict {
                                 last_seat_conflict = error
                                     .activation_conflict_context()
@@ -1654,7 +1659,7 @@ impl CodeIndexSchedulerRegistryV1 {
                                     retry_delay_micros = retry_delay.as_micros() as u64,
                                     error = %error,
                                     "graph activation failed retryably; the sealed generation \
-                                     stays unseated until the scheduled retry"
+                                     still seats and the next pass retries native graph"
                                 );
                                 hotpath::gauge!("daemon.code_index.graph_seat.retry_total")
                                     .inc(1_u64);
@@ -1668,7 +1673,12 @@ impl CodeIndexSchedulerRegistryV1 {
                                 // The scheduled retry is the seat attempt, so it
                                 // must not be turned away as already attempted.
                                 graph_seat_attempted = None;
-                                result = Ok((Err(error), None, None));
+                                if !super::activation_failure_keeps_serving_candidate(
+                                    error.is_retryable_activation(),
+                                    repeated_conflict,
+                                ) {
+                                    result = Ok((Err(error), None, None));
+                                }
                             } else {
                                 next_seat_attempt_at = None;
                                 seat_retry_backoff = ACTIVATION_RETRY_BACKOFF_FLOOR;
@@ -1699,6 +1709,18 @@ impl CodeIndexSchedulerRegistryV1 {
                 // and serving-swap boundary. Graph work above ran only when
                 // the outcome was ready.
                 if let Some(outcome) = published_text_projection_outcome.take() {
+                    // A clone-fingerprint successor is still `Unfinished` work
+                    // after exact and lexical owners are ready. That must not
+                    // clear the prepared generation the way a missing owner does.
+                    let owners_ready = exact_and_lexical_ready_for_graph(graph_text.as_ref());
+                    let outcome = match outcome {
+                        PublishedTextProjectionOutcomeV1::Unfinished
+                            if !super::text_projection_unfinished_withholds_seat(owners_ready) =>
+                        {
+                            PublishedTextProjectionOutcomeV1::Finished
+                        }
+                        other => other,
+                    };
                     match outcome {
                         PublishedTextProjectionOutcomeV1::Finished => {
                             // The seat needs only the ready exact/lexical