22
33from __future__ import annotations
44
5- import functools
65import json
76import logging
7+ import os
88import threading
99from pathlib import Path
1010from typing import Any , TypedDict
1414BASELINE_PATH = Path (__file__ ).resolve ().parent .parent / "schema_baseline.json"
1515
1616_lock = threading .Lock ()
17- # Accumulated drift from parse_session() runs in this process; cleared only via
18- # reset_schema_report() (tests) or server restart.
17+ # Process-wide union of *new* field paths seen since server start (or reset_schema_report).
18+ # Intentionally sticky: once upstream drift is detected, the banner stays until restart so
19+ # operators do not miss it. missing_fields reflects only the most recent sampled parse.
20+ _baseline_cache : tuple [frozenset [str ], frozenset [str ]] | None = None
21+ _baseline_load_failed : bool = False
1922_last_report : SchemaDriftReport = {
2023 "known_fields" : [],
2124 "new_fields" : [],
@@ -31,6 +34,21 @@ class SchemaDriftReport(TypedDict):
3134 has_drift : bool
3235
3336
37+ def is_schema_drift_enabled () -> bool :
38+ """Return False when CLAUDE_CODE_CHAT_BROWSER_SCHEMA_DRIFT=0|false|no."""
39+ flag = os .environ .get ("CLAUDE_CODE_CHAT_BROWSER_SCHEMA_DRIFT" , "1" ).strip ().lower ()
40+ return flag not in "0" , "false" , "no" )
41+
42+
43+ def schema_drift_sample_limit () -> int :
44+ """Max JSONL records per session to fingerprint (default 3). Set 0 to disable sampling cap."""
45+ raw = os .environ .get ("CLAUDE_CODE_CHAT_BROWSER_SCHEMA_DRIFT_SAMPLE" , "3" ).strip ()
46+ try :
47+ return max (0 , int (raw ))
48+ except ValueError :
49+ return 3
50+
51+
3452def collect_field_paths (record : dict [str , Any ], prefix : str = "" ) -> set [str ]:
3553 """Recursively collect dotted JSON paths (with ``[]`` for list items)."""
3654 paths : set [str ] = set ()
@@ -48,33 +66,44 @@ def collect_field_paths(record: dict[str, Any], prefix: str = "") -> set[str]:
4866 return paths
4967
5068
51- @functools .lru_cache (maxsize = 1 )
52- def _cached_baseline () -> tuple [frozenset [str ], frozenset [str ]]:
53- """Load and cache known/required field paths from ``schema_baseline.json``."""
54- raw = json .loads (BASELINE_PATH .read_text (encoding = "utf-8" ))
55- fields = raw .get ("fields" , {})
56- if not isinstance (fields , dict ):
57- raise ValueError ("schema_baseline.json: 'fields' must be an object" )
58- known_paths : set [str ] = set ()
59- required_paths : set [str ] = set ()
60- for path , spec in fields .items ():
61- if not isinstance (path , str ):
62- continue
63- known_paths .add (path )
64- if isinstance (spec , dict ) and spec .get ("required" ):
65- required_paths .add (path )
66- return frozenset (known_paths ), frozenset (required_paths )
69+ def _load_baseline () -> tuple [frozenset [str ], frozenset [str ]]:
70+ """Load baseline paths once; cache success and remember hard failures."""
71+ global _baseline_cache , _baseline_load_failed
72+ if _baseline_load_failed :
73+ raise ValueError ("schema_baseline.json previously failed to load" )
74+ if _baseline_cache is not None :
75+ return _baseline_cache
76+ try :
77+ raw = json .loads (BASELINE_PATH .read_text (encoding = "utf-8" ))
78+ fields = raw .get ("fields" , {})
79+ if not isinstance (fields , dict ):
80+ raise ValueError ("schema_baseline.json: 'fields' must be an object" )
81+ known_paths : set [str ] = set ()
82+ required_paths : set [str ] = set ()
83+ for path , spec in fields .items ():
84+ if not isinstance (path , str ):
85+ continue
86+ known_paths .add (path )
87+ if isinstance (spec , dict ) and spec .get ("required" ):
88+ required_paths .add (path )
89+ _baseline_cache = (frozenset (known_paths ), frozenset (required_paths ))
90+ return _baseline_cache
91+ except (OSError , json .JSONDecodeError , ValueError , TypeError ) as exc :
92+ if not _baseline_load_failed :
93+ _baseline_load_failed = True
94+ _log .warning ("schema drift baseline load failed (will not retry): %s" , exc )
95+ raise
6796
6897
6998def load_baseline_fields () -> dict [str , bool ]:
7099 """Return baseline field paths mapped to whether each path is required."""
71- known_paths , required_paths = _cached_baseline ()
100+ known_paths , required_paths = _load_baseline ()
72101 return {path : path in required_paths for path in known_paths }
73102
74103
75104def diff_against_baseline (observed_paths : set [str ]) -> SchemaDriftReport :
76- """Compare observed session field paths to the committed baseline."""
77- known_paths , required_paths = _cached_baseline ()
105+ """Compare observed session field paths to the committed baseline (paths only, not types) ."""
106+ known_paths , required_paths = _load_baseline ()
78107 known_fields = sorted (known_paths )
79108 new_fields = sorted (observed_paths - known_paths )
80109 missing_fields = sorted (required_paths - observed_paths )
@@ -90,29 +119,32 @@ def record_parse_drift(observed_paths: set[str]) -> SchemaDriftReport | None:
90119 """Diff *observed_paths*, log warnings, and merge into the process-wide report."""
91120 try :
92121 report = diff_against_baseline (observed_paths )
93- except (OSError , json .JSONDecodeError , ValueError , TypeError ) as exc :
94- _log .warning ("schema drift tracking skipped: %s" , exc )
122+ except (OSError , json .JSONDecodeError , ValueError , TypeError ):
95123 return None
96124
97- if report ["new_fields" ]:
125+ with _lock :
126+ global _last_report
127+ prior_new = set (_last_report ["new_fields" ])
128+ genuinely_new = sorted (set (report ["new_fields" ]) - prior_new )
129+ merged_new = sorted (prior_new | set (report ["new_fields" ]))
130+
131+ if genuinely_new :
98132 _log .warning (
99133 "schema drift: new JSONL field paths not in baseline: %s" ,
100- report [ "new_fields" ] ,
134+ genuinely_new ,
101135 )
102136 if report ["missing_fields" ]:
103137 _log .warning (
104- "schema drift: missing required JSONL field paths: %s" ,
138+ "schema drift: missing required JSONL field paths in sampled records : %s" ,
105139 report ["missing_fields" ],
106140 )
141+
107142 with _lock :
108- global _last_report
109- merged_new = sorted (set (_last_report ["new_fields" ]) | set (report ["new_fields" ]))
110- merged_missing = sorted (set (_last_report ["missing_fields" ]) | set (report ["missing_fields" ]))
111143 _last_report = {
112144 "known_fields" : report ["known_fields" ],
113145 "new_fields" : merged_new ,
114- "missing_fields" : merged_missing ,
115- "has_drift" : bool (merged_new or merged_missing ),
146+ "missing_fields" : list ( report [ "missing_fields" ]) ,
147+ "has_drift" : bool (merged_new or report [ "missing_fields" ] ),
116148 }
117149 return report
118150
@@ -142,4 +174,6 @@ def reset_schema_report() -> None:
142174
143175def clear_baseline_cache () -> None :
144176 """Clear the cached baseline (for tests)."""
145- _cached_baseline .cache_clear ()
177+ global _baseline_cache , _baseline_load_failed
178+ _baseline_cache = None
179+ _baseline_load_failed = False
0 commit comments