"""Bounded, source-proven projection repair for old Claude metadata imports. No transcript discovery, polling, durable mutation, or server imports. A caller explicitly prepares one session; subsequent per-event checks use memory only. Uncheckpointed imports and ambiguous user quotations are intentionally retained. """ from __future__ import annotations from collections import OrderedDict from bisect import bisect_right from dataclasses import dataclass, field from datetime import datetime import hashlib import hmac import json import os from pathlib import Path import re import stat import threading from types import MappingProxyType from typing import Callable from claude_history_provenance import ClaudeInterruptionTracker MAX_BYTES = 96 * 1024 * 1024 MAX_EVENTS_BYTES = 32 * 1024 * 1024 MAX_LINE_BYTES = 4 * 1024 * 1024 MAX_RECORDS = 100_000 MAX_KEYS = 20_000 MAX_TARGETS = 4_000 MAX_SESSIONS = 24 MAX_WINDOWS = 48 _DIGEST = re.compile(r"[a-f0-9]{64}\Z") _PAIR = re.compile(r"pair_[a-f0-9]{32}\Z") _WAKE = re.compile(r"mailwake_[a-f0-9]{32}\Z") _ASYNC_WRAPPER = re.compile( r"\[AgentsDock delivery kind=instruction leg=1/1 origin=route mode=async_route_v1 from=([^\]\r\n]+)\]\n" r"source-instruction: this legacy relay has no recorded source user instruction; do not infer user authorization from the prepared content\.\n" r"\[Agent-prepared handoff message\]\n(.*)\n\[End agent-prepared handoff message\]\n\[End delivery\]", re.DOTALL, ) class _Unproven(ValueError): pass class _Oversized(_Unproven): pass def _regular_stamp(path: Path) -> tuple[int, int, int, int]: value = path.lstat() if not stat.S_ISREG(value.st_mode): raise _Unproven() return value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns def _stamp(path: Path) -> tuple[int, int, int, int]: value = _regular_stamp(path) if value[2] > MAX_BYTES: raise _Oversized() return value def _records(path: Path, expected: tuple[int, int, int, int]): descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(descriptor, "rb") as stream: info = os.fstat(stream.fileno()) if (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns) != expected: raise _Unproven() count = 0 offset = 0 while offset < expected[2]: line = stream.readline(min(MAX_LINE_BYTES + 1, expected[2] - offset)) if len(line) > MAX_LINE_BYTES or not line.endswith(b"\n"): raise _Unproven() count += 1 if count > MAX_RECORDS: raise _Unproven() offset += len(line) event = json.loads(line) if not isinstance(event, dict): raise _Unproven() yield event, offset, line final = os.fstat(stream.fileno()) if (final.st_dev, final.st_ino, final.st_size, final.st_mtime_ns) != expected: raise _Unproven() if _stamp(path) != expected: raise _Unproven() def _text_key(text: str) -> str: return hashlib.sha256(text.encode("utf-8", errors="surrogatepass")).hexdigest() def _target(event: dict) -> tuple[int, str, str] | None: seq = event.get("seq") run = event.get("run_id") prompt = event.get("prompt") if ( event.get("type") != "turn_started" or event.get("imported") is not True or event.get("backend") != "claude" or type(seq) is not int or seq <= 0 or not isinstance(run, str) or not run.startswith("import_") or not isinstance(prompt, str) or len(prompt) > MAX_LINE_BYTES ): return None return seq, run, _text_key(prompt) def _assistant_target(event: dict) -> tuple | None: origin = event.get("provider_origin") text, seq, run = event.get("text"), event.get("seq"), event.get("run_id") if (event.get("type") not in ("assistant_text", "reasoning_summary") or event.get("imported") is not True or event.get("backend") != "claude" or type(seq) is not int or seq <= 0 or not isinstance(run, str) or not run.startswith("import_") or not isinstance(text, str) or not text.strip() or len(text) > MAX_LINE_BYTES or not isinstance(origin, dict) or origin.get("provider") != "claude" or origin.get("kind") not in (None, "assistant") or event.get("provider_user_authored") is True or any(event.get(key) for key in ("clientUserMessageId", "clientId", "client_user_message_id", "client_id"))): return None identity = tuple(origin.get(key) for key in ("event_id", "session_id", "timestamp")) if (not all(isinstance(value, str) and 0 < len(value) <= 256 for value in identity) or _timestamp(identity[2]) is None): return None return event["type"], seq, run, _text_key(text.strip()), identity def _async_delivery_identity(event: dict) -> tuple | None: pair, message = event.get("conversation_id"), event.get("message_id") source, target = event.get("source_session_id"), event.get("target_session_id") if (event.get("conversation_mode") != "async_route_v1" or not isinstance(pair, str) or not _PAIR.fullmatch(pair) or not isinstance(message, str) or not message or message != event.get("cross_chat_envelope_id") or not isinstance(source, str) or not source or not isinstance(target, str) or not target or source == target or event.get("session_id") != target): return None return pair, message, source, target def _mailbox_wake_identity(event: dict) -> tuple | None: claim, cutoff, digest = (event.get(key) for key in ( "mailbox_wake_id", "mailbox_wake_through_seq", "provider_input_sha256")) if (event.get("purpose") != "chat_mailbox_wake" or event.get("provider_generated") is not True or event.get("prompt") != "" or event.get("provider_user_authored") is True or any(event.get(key) for key in ("clientUserMessageId", "clientId", "client_user_message_id", "client_id")) or not isinstance(claim, str) or not _WAKE.fullmatch(claim) or type(cutoff) is not int or not 0 < cutoff < 2**63 or not isinstance(digest, str) or not _DIGEST.fullmatch(digest)): return None return claim, cutoff, digest class _AssistantReplays: """Bounded exact source/output correlation, fed by the existing two reads.""" def __init__(self, provider_id: str, normalize_assistant: Callable[[str], str] | None = None) -> None: self.provider_id = provider_id self.normalize_assistant = normalize_assistant self.starts, self.ends, self.owners, self.native = {}, {}, {}, [] self.candidates, self.sources, self.identities, self.source_credits = [], {}, {}, {} self.source_display, self.source_message_counts = {}, {} def display_key(self, text: str) -> str | None: if self.normalize_assistant is None: return None display = self.normalize_assistant(text) return (_text_key(display) if isinstance(display, str) and display.strip() and len(display) <= MAX_LINE_BYTES else None) def event(self, event: dict) -> None: target = _assistant_target(event) if target is not None and target[4][1] == self.provider_id: self.candidates.append((target, event.get("phase"))) run = event.get("run_id") if (isinstance(run, str) and run and not run.startswith("import_") and event.get("backend") == "claude" and event.get("imported") is not True): kind = event.get("type") if kind == "turn_started": self.starts.setdefault(run, []).append(event) elif kind == "turn_finished": self.ends.setdefault(run, []).append(event) elif kind == "provider_session": self.owners.setdefault(run, []).append(event) elif (kind in ("reasoning_summary", "assistant_text") and (event.get("phase") in ("commentary", "final_answer") or (kind == "assistant_text" and event.get("phase") is None)) and isinstance(event.get("text"), str) and event["text"].strip() and len(event["text"]) <= MAX_LINE_BYTES): self.native.append((event, _text_key(event["text"].strip()))) if sum(map(len, (self.starts, self.ends, self.owners, self.native, self.candidates))) > MAX_TARGETS: raise _Unproven() def source(self, event: dict, offset: int) -> None: if event.get("type") != "assistant" or event.get("sessionId") != self.provider_id: return identity = (event.get("uuid"), self.provider_id, event.get("timestamp")) if not all(isinstance(value, str) and 0 < len(value) <= 256 for value in identity): return self.identities[identity] = self.identities.get(identity, 0) + 1 self.source_message_counts[identity[0]] = self.source_message_counts.get(identity[0], 0) + 1 if event.get("isSidechain") is True or event.get("isMeta") is True: return message = event.get("message") content = message.get("content") if isinstance(message, dict) else None # Only public text blocks. Thinking/tool payloads cannot establish a # public replay, even if their wording happens to match another row. if isinstance(content, str): text = content elif isinstance(content, list): text = "\n".join(block["text"] for block in content if isinstance(block, dict) and block.get("type") == "text" and isinstance(block.get("text"), str)) else: return if text.strip() and len(text) <= MAX_LINE_BYTES: digest = _text_key(text.strip()) self.sources.setdefault((identity, digest), []).append(offset) self.source_display[(identity, digest)] = self.display_key(text) timestamp = _timestamp(identity[2]) if timestamp is not None: self.source_credits.setdefault((int(timestamp), digest), set()).add(identity) if len(self.identities) + len(self.sources) > MAX_KEYS: raise _Unproven() def prove(self, eligible: dict) -> frozenset: native, identified = {}, {} for event, digest in self.native: run = event["run_id"] starts, ends = self.starts.get(run, ()), self.ends.get(run, ()) if len(starts) != 1 or len(ends) != 1: continue start, end = starts[0], ends[0] owners = (start, event, end, *self.owners.get(run, ())) provider_ids = {row.get("provider_session_id") for row in owners if row.get("provider_session_id") not in (None, "")} delivery = _async_delivery_identity(start) scheduled = start.get("purpose") == "scheduled_job" and bool(start.get("job_id")) async_delivery = (start.get("purpose") == "cross_chat_handoff_delivery" and delivery and _async_delivery_identity(end) == delivery and _async_delivery_identity(event) == delivery and isinstance(event.get("provider_message_id"), str) and event["provider_message_id"]) mailbox_wake = (_mailbox_wake_identity(start) and isinstance(event.get("provider_message_id"), str) and event["provider_message_id"]) if (not (scheduled or async_delivery or mailbox_wake) or provider_ids != {self.provider_id} or end.get("exit_code") != 0 or end.get("stopped") is True or any(type(row.get("seq")) is not int or not start["seq"] <= row["seq"] <= end["seq"] for row in self.owners.get(run, ())) or (scheduled and event.get("job_id") not in (None, "", start["job_id"]))): continue if (event.get("phase") is None and (not isinstance(end.get("result_text"), str) or _text_key(end["result_text"].strip()) != digest)): continue times = [_timestamp(row.get("ts")) for row in (start, event, end)] seqs = [row.get("seq") for row in (start, event, end)] if (any(value is None for value in times) or not times[0] <= times[1] <= times[2] or not all(type(value) is int for value in seqs) or not seqs[0] < seqs[1] < seqs[2]): continue native.setdefault((int(times[1]), digest), []).append(event) message_id = event.get("provider_message_id") if isinstance(message_id, str) and 0 < len(message_id) <= 256: identified.setdefault(message_id, []).append((event, digest)) counts = {} for target, _phase in self.candidates: key = (target[2], target[4]) counts[key] = counts.get(key, 0) + 1 proven = set() for target, phase in self.candidates: _kind, seq, run, digest, identity = target batch = eligible.get(run) matches = native.get((int(_timestamp(identity[2])), digest), ()) offsets = self.sources.get((identity, digest), ()) owned = identified.get(identity[0], ()) if owned: # Prefer the immutable message owner over text/time buckets. # A normalized match is permitted ONLY for one native UUID # and one exact raw source record. The imported raw digest, # checkpoint, phase and timestamp checks remain authoritative. # Never use decoration equivalence for identityless messages. matches = () if len(owned) == 1 and self.source_message_counts.get(identity[0]) == 1: event, native_digest = owned[0] canonical = self.source_display.get((identity, digest)) if (int(_timestamp(event.get("ts"))) == int(_timestamp(identity[2])) and (native_digest == digest or (canonical and canonical == self.display_key(event["text"])))): matches = (event,) if (batch is None or len(matches) != 1 or len(offsets) != 1 or self.identities.get(identity) != 1 or counts[(run, identity)] != 1): continue match = matches[0] expected_phase = match.get("phase") or "final_answer" if (phase not in (None, expected_phase) or match.get("provider_message_id") not in (None, "", identity[0])): continue source_time, native_time = _timestamp(identity[2]), _timestamp(match.get("ts")) if ("." in match["ts"] and source_time != native_time): continue # One coarse native timestamp is not credit for two distinct # source messages. An exact native message UUID may disambiguate. if (not match.get("provider_message_id") and len(self.source_credits.get((int(source_time), digest), ())) != 1): continue first, last, start, end = batch[:4] if first < seq < last and start < offsets[0] <= end: proven.add(target) return frozenset(proven) class _AsyncDeliveryInputs: """Exact completed delivery receipt + complete source wrapper, not text guessing.""" def __init__(self, provider_id, native, normalize_user, normalize_full_user): self.provider_id, self.native = provider_id, native self.normalize_user, self.normalize_full_user = normalize_user, normalize_full_user self.receipts, self.candidates, self.sources, self.source_counts = {}, [], {}, {} def event(self, event): if event.get("type") == "chat_conversation_message_started" and _async_delivery_identity(event): run = event.get("target_run_id") if isinstance(run, str) and run and not run.startswith("import_"): self.receipts.setdefault(run, []).append(event) target, origin = _target(event), event.get("provider_origin") if (target and isinstance(origin, dict) and origin.get("provider") == "claude" and origin.get("kind") in (None, "user") and origin.get("session_id") == self.provider_id and event.get("provider_user_authored") is not True and not any(event.get(key) for key in ("clientUserMessageId", "clientId", "client_user_message_id", "client_id"))): identity = tuple(origin.get(key) for key in ("event_id", "session_id", "timestamp")) if all(isinstance(value, str) and 0 < len(value) <= 256 for value in identity): self.candidates.append((target, identity)) if len(self.receipts) + len(self.candidates) > MAX_TARGETS: raise _Unproven() def source(self, event, offset): if event.get("type") != "user" or event.get("sessionId") != self.provider_id or self.normalize_full_user is None: return identity = (event.get("uuid"), self.provider_id, event.get("timestamp")) if not all(isinstance(value, str) and 0 < len(value) <= 256 for value in identity): return self.source_counts[identity] = self.source_counts.get(identity, 0) + 1 if (any(event.get(key) is True for key in ("isMeta", "isCompactSummary", "isSidechain")) or any(event.get(key) for key in ("clientUserMessageId", "clientId", "client_user_message_id", "client_id"))): return full, display = self.normalize_full_user(event), self.normalize_user(event) match = _ASYNC_WRAPPER.fullmatch(full.strip()) if isinstance(full, str) and len(full) <= MAX_LINE_BYTES else None timestamp = _timestamp(identity[2]) if match and isinstance(display, str) and display and timestamp is not None: self.sources.setdefault((match[1], _text_key(match[2]), len(match[2]), int(timestamp)), []).append( (identity, _text_key(display), offset)) if len(self.source_counts) + len(self.sources) > MAX_KEYS: raise _Unproven() def prove(self, eligible): matches, credits, candidate_counts = {}, {}, {} for target, identity in self.candidates: candidate_counts[(target[1], identity)] = candidate_counts.get((target[1], identity), 0) + 1 for run, receipts in self.receipts.items(): starts, ends = self.native.starts.get(run, ()), self.native.ends.get(run, ()) if len(receipts) != 1 or len(starts) != 1 or len(ends) != 1: continue start, end, receipt = starts[0], ends[0], receipts[0] delivery = _async_delivery_identity(start) owners = (start, end, *self.native.owners.get(run, ())) provider_ids = {row.get("provider_session_id") for row in owners if row.get("provider_session_id") not in (None, "")} start_time, end_time = _timestamp(start.get("ts")), _timestamp(end.get("ts")) if (start.get("purpose") != "cross_chat_handoff_delivery" or not delivery or _async_delivery_identity(end) != delivery or _async_delivery_identity(receipt) != delivery or receipt.get("handoff_authorization_kind") != "configured_route" or receipt.get("handoff_status") != "running" or receipt.get("queued_id") != start.get("queued_id") or not isinstance(receipt.get("handoff_body_sha256"), str) or not _DIGEST.fullmatch(receipt["handoff_body_sha256"]) or type(receipt.get("handoff_body_chars")) is not int or provider_ids != {self.provider_id} or end.get("exit_code") != 0 or end.get("stopped") is True or start_time is None or end_time is None or start_time > end_time or any(type(row.get("seq")) is not int for row in (start, receipt, end)) or not start["seq"] < receipt["seq"] < end["seq"] or any(type(row.get("seq")) is not int or not start["seq"] <= row["seq"] <= end["seq"] for row in self.native.owners.get(run, ()))): continue source_key = (receipt.get("source_title"), receipt["handoff_body_sha256"], receipt["handoff_body_chars"], int(start_time)) for identity, key, offset in self.sources.get(source_key, ()): timestamp = _timestamp(identity[2]) if (self.source_counts.get(identity) != 1 or timestamp is None or not start_time <= timestamp <= end_time or int(timestamp) != int(start_time) or ("." in start["ts"] and timestamp != start_time)): continue matches.setdefault((identity, key), []).append((run, offset)) credits[run] = credits.get(run, 0) + 1 targets = set() for target, identity in self.candidates: seq, run, key = target batch, matched = eligible.get(run), matches.get((identity, key), ()) if batch and len(matched) == 1 and candidate_counts[(run, identity)] == 1 and credits.get(matched[0][0]) == 1: first, last, start, end = batch[:4] if first < seq < last and start < matched[0][1] <= end: targets.add(target) return targets class _MailboxWakeInputs(_AsyncDeliveryInputs): """Exact generated native input hash, current provider and checkpoint identity.""" def __init__(self, *args): super().__init__(*args) self.wake_hashes = None def source(self, event, offset): if self.wake_hashes is None: self.wake_hashes = {wake[2] for starts in self.native.starts.values() for start in starts if (wake := _mailbox_wake_identity(start)) is not None} if not self.wake_hashes: return if event.get("type") != "user" or event.get("sessionId") != self.provider_id or self.normalize_full_user is None: return identity = (event.get("uuid"), self.provider_id, event.get("timestamp")) if not all(isinstance(value, str) and 0 < len(value) <= 256 for value in identity): return self.source_counts[identity] = self.source_counts.get(identity, 0) + 1 if len(self.source_counts) > MAX_KEYS: raise _Unproven() if (any(event.get(key) is True for key in ("isMeta", "isCompactSummary", "isSidechain", "provider_user_authored")) or any(event.get(key) for key in ("clientUserMessageId", "clientId", "client_user_message_id", "client_id"))): return full, display = self.normalize_full_user(event), self.normalize_user(event) timestamp = _timestamp(identity[2]) if (isinstance(full, str) and full and len(full) <= MAX_LINE_BYTES and _text_key(full) in self.wake_hashes and isinstance(display, str) and display and timestamp is not None): self.sources.setdefault(_text_key(full), []).append( (identity, _text_key(display), offset)) if len(self.source_counts) + len(self.sources) > MAX_KEYS: raise _Unproven() def prove(self, eligible): matches, credits, candidate_counts = {}, {}, {} for target, identity in self.candidates: key = (target[1], identity) candidate_counts[key] = candidate_counts.get(key, 0) + 1 for run, starts in self.native.starts.items(): ends = self.native.ends.get(run, ()) if len(starts) != 1 or len(ends) != 1: continue start, end = starts[0], ends[0] wake = _mailbox_wake_identity(start) owners = (start, end, *self.native.owners.get(run, ())) provider_ids = {row.get("provider_session_id") for row in owners if row.get("provider_session_id") not in (None, "")} start_time, end_time = _timestamp(start.get("ts")), _timestamp(end.get("ts")) if (wake is None or provider_ids != {self.provider_id} or end.get("exit_code") != 0 or end.get("stopped") is True or start_time is None or end_time is None or start_time > end_time or type(start.get("seq")) is not int or type(end.get("seq")) is not int or start["seq"] >= end["seq"] or any(type(row.get("seq")) is not int or not start["seq"] <= row["seq"] <= end["seq"] for row in self.native.owners.get(run, ()))): continue for identity, key, offset in self.sources.get(wake[2], ()): timestamp = _timestamp(identity[2]) if (self.source_counts.get(identity) != 1 or timestamp is None or not start_time <= timestamp <= end_time): continue matches.setdefault((identity, key), []).append((run, offset)) credits[run] = credits.get(run, 0) + 1 targets = set() for target, identity in self.candidates: seq, run, key = target batch, matched = eligible.get(run), matches.get((identity, key), ()) if batch and len(matched) == 1 and candidate_counts[(run, identity)] == 1 and credits.get(matched[0][0]) == 1: first, last, start, end = batch[:4] if first < seq < last and start < matched[0][1] <= end: targets.add(target) return targets def filter_native_claude_mailbox_wake_items( session_id: str, provider_id: str, events: Path, items: list[dict], *, sync_checkpoint: dict, root: Path, normalize_user: Callable, normalize_full_user: Callable, source_path: Path | None = None, normalize_assistant: Callable[[str], str] | None = None, ) -> list[dict]: """First-import wake input/output proof; never a text-only filter. Parsed items omit human/client flags and may collapse repeated text, so the exact checkpoint source is required to prove unique nonhuman ownership. All failures retain the original items. Native output is never removed. """ try: if not items or len(items) > MAX_TARGETS: return items stamp = _regular_stamp(events) native = _AssistantReplays(provider_id, normalize_assistant) for event, _offset in _bounded_records(events, stamp, max(0, stamp[2] - MAX_EVENTS_BYTES), stamp[2]): if (event.get("session_id") in (None, "", session_id) and (event.get("run_id") in native.starts or _mailbox_wake_identity(event))): native.event(event) if not any(_mailbox_wake_identity(start) for starts in native.starts.values() for start in starts): return items cursor = sync_checkpoint.get("cursor") if (sync_checkpoint.get("version") != 1 or not isinstance(cursor, dict) or cursor.get("version") != 1 or cursor.get("backend") != "claude" or cursor.get("provider_session_id") != provider_id): return items path = cursor.get("source_path") if not isinstance(path, str): return items source = Path(path) if source_path is not None and source != source_path: return items if not source.is_absolute() or source.suffix != ".jsonl" or source.stem != provider_id or source.is_symlink(): return items source = source.resolve(strict=True) source.relative_to(root.resolve(strict=True)) source_stamp = _regular_stamp(source) start, end = sync_checkpoint.get("previous_source_offset"), cursor.get("source_offset") previous, expected = sync_checkpoint.get("previous_source_digest"), cursor.get("source_digest") if (type(start) is not int or type(end) is not int or not 0 <= start < end <= source_stamp[2] or (cursor.get("source_dev"), cursor.get("source_ino")) != source_stamp[:2] or type(sync_checkpoint.get("previous_present")) is not bool or not isinstance(expected, str) or not _DIGEST.fullmatch(expected) or start == 0 and previous != "" or start > 0 and (sync_checkpoint["previous_present"] is not True or not isinstance(previous, str) or not _DIGEST.fullmatch(previous))): return items proof = _MailboxWakeInputs(provider_id, native, normalize_user, normalize_full_user) # Temporary proof coordinates only; no synthetic event is persisted. batch = "import_mailbox_wake_filter" assistant_targets = {} for index, item in enumerate(items, 1): if item.get("kind") == "user" and item.get("source_text_sha256") is None: proof.event({**item, "seq": index, "run_id": batch, "type": "turn_started", "backend": "claude", "imported": True, "prompt": item.get("text")}) elif item.get("kind") == "assistant": candidate = {**item, "seq": index, "run_id": batch, "type": "assistant_text", "backend": "claude", "imported": True} native.event(candidate) assistant_targets[index] = _assistant_target(candidate) if source_stamp[2] <= MAX_EVENTS_BYTES: digest, verified = hashlib.sha256(), set() for event, offset, line in _records(source, source_stamp): digest.update(line) if offset == start and hmac.compare_digest(digest.hexdigest(), previous): verified.add(start) if offset == end and hmac.compare_digest(digest.hexdigest(), expected): verified.add(end) if offset <= end: proof.source(event, offset) native.source(event, offset) if end not in verified or start and start not in verified: return items else: # Same bounded proof as historical semantic windows: the caller has # just verified this checkpoint. Recheck file identity and every # candidate's exact source UUID/full hash, without a prefix rescan. window_start = max(0, end - MAX_EVENTS_BYTES) if start < window_start: return items for event, offset in _bounded_records(source, source_stamp, window_start, end): proof.source(event, offset) native.source(event, offset) eligible = {batch: (0, len(items) + 1, start, end)} targets, assistant_replays = proof.prove(eligible), native.prove(eligible) projected = [] for index, item in enumerate(items, 1): reason = ("source_proven_import" if (index, batch, _text_key(item.get("text", ""))) in targets else "source_proven_assistant_replay" if assistant_targets.get(index) in assistant_replays else None) projected.append({**item, "text": "", "metadata_only": True, "provider_history_repair": reason} if reason else item) return projected except (OSError, ValueError, TypeError, KeyError, RuntimeError): return items @dataclass(frozen=True) class _Proof: provider_id: str events_stamp: tuple[int, int, int, int] source: Path | None source_stamp: tuple[int, int, int, int] | None targets: frozenset[tuple[int, str, str]] interruptions: tuple[tuple[tuple[int, str, str], str], ...] = () companions: frozenset[tuple[str, int, str]] = frozenset() assistant_replays: frozenset[tuple] = frozenset() interruption_index: MappingProxyType = field(init=False, repr=False) cache_signature: frozenset = field(init=False, repr=False) def __post_init__(self) -> None: object.__setattr__(self, "interruption_index", MappingProxyType(dict(self.interruptions))) object.__setattr__(self, "cache_signature", self.targets | frozenset( ("interruption", key, origin) for key, origin in self.interruptions) | frozenset(("companion", *key) for key in self.companions) | frozenset(("assistant_replay", *key) for key in self.assistant_replays)) def signature(self) -> frozenset: return self.cache_signature def _timestamp(value) -> float | None: if not isinstance(value, str) or len(value) > 64: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.timestamp() if parsed.tzinfo is not None else None except (ValueError, OverflowError, OSError): return None def _steer_intervals(events: list[dict], provider_id: str) -> list[tuple[float, float]]: """Only a complete native handoff can establish why a provider was interrupted.""" queues, finished, starts = {}, {}, {} for event in events: kind = event.get("type") if kind == "turn_queue_run_now": key = event.get("interrupted_run_id") destination = queues elif kind in ("turn_finished", "turn_stopped"): key = event.get("run_id") destination = finished else: key = event.get("steer_interrupted_run_id") destination = starts if isinstance(key, str) and key and not key.startswith("import_"): destination.setdefault(key, []).append(event) intervals = [] for run, queued in queues.items(): ended, following = finished.get(run, ()), starts.get(run, ()) if len(queued) != 1 or len(ended) != 1 or len(following) != 1: continue queue, terminal, start = queued[0], ended[0], following[0] queued_id = queue.get("queued_id") next_run = start.get("run_id") terminal_proven = ( terminal.get("type") == "turn_finished" and terminal.get("stopped") is True and terminal.get("exit_code") is None ) or ( terminal.get("type") == "turn_stopped" and terminal.get("native_steer") is True and terminal.get("superseded_by_run_id") == next_run ) if ( not terminal_proven or terminal.get("provider_session_id") != provider_id or not isinstance(queued_id, str) or not queued_id or start.get("queued_id") != queued_id or not isinstance(next_run, str) or not next_run or next_run == run or next_run.startswith("import_") or any(event.get("provider_session_id") not in (None, "", provider_id) for event in (queue, start)) ): continue times = [_timestamp(event.get("ts")) for event in (queue, terminal, start)] if any(value is None for value in times): continue # An interruption must be within five seconds of every corroborating # record, not merely near an unrelated stop elsewhere in this chat. lower, upper = max(times) - 5.0, min(times) + 5.0 if lower <= upper: intervals.append((lower, upper)) return intervals def _interruption_origin(origin: dict, provider_id: str, intervals) -> dict | None: if ( not isinstance(origin, dict) or origin.get("provider") != "claude" or origin.get("kind") != "interruption" or origin.get("session_id") != provider_id or not isinstance(origin.get("event_id"), str) ): return None timestamp = _timestamp(origin.get("timestamp")) if timestamp is None: return None result = {key: origin[key] for key in ( "provider", "kind", "event_id", "session_id", "timestamp", "parent_event_id", "prompt_id", ) if key in origin} result["cause"] = "steer" if any(low <= timestamp <= high for low, high in intervals) else "unknown" return result def _native_control(event: dict, session_id: str) -> dict | None: if not ( event.get("session_id") == session_id and event.get("backend") == "claude" and event.get("imported") is not True and event.get("type") in ("turn_queue_run_now", "turn_finished", "turn_stopped", "turn_started") ): return None return {key: event[key] for key in ( "type", "ts", "run_id", "queued_id", "interrupted_run_id", "steer_interrupted_run_id", "provider_session_id", "stopped", "exit_code", "native_steer", "superseded_by_run_id", ) if key in event} def enrich_interruption_origins(events_path: Path, provider_id: str, origins: list[dict], *, session_id: str) -> list[dict]: """Explicit import-worker enrichment; bounded native-log read, no source discovery. Valid origins retain their order and are copied with unknown cause on any read/proof failure. Invalid entries become empty dictionaries; an oversized input returns an empty list. Callers must not replace an origin with either. """ if not isinstance(origins, list) or len(origins) > MAX_TARGETS: return [] unknown = [_interruption_origin(origin, provider_id, ()) or {} for origin in origins] if not any(unknown) or not isinstance(session_id, str) or not session_id: return unknown try: stamp = _stamp(events_path) if stamp[2] > MAX_EVENTS_BYTES: raise _Unproven() controls = [] for event, _offset, _line in _records(events_path, stamp): control = _native_control(event, session_id) if control is not None: controls.append(control) if len(controls) > MAX_TARGETS: raise _Unproven() intervals = _steer_intervals(controls, provider_id) return [_interruption_origin(origin, provider_id, intervals) or {} for origin in unknown] except (OSError, ValueError, TypeError, KeyError, RuntimeError): return unknown def _prove(session_id: str, provider_id: str, events: Path, root: Path, normalize_user: Callable[[dict], str | None], events_stamp, normalize_full_user: Callable[[dict], str | None] | None = None, normalize_assistant: Callable[[str], str] | None = None) -> _Proof: empty = _Proof(provider_id, events_stamp, None, None, frozenset()) batches = {} candidates = [] terminals = {} run_rows = {} terminal_counts = {} native_events = [] scheduled_starts, scheduled_ends, candidate_origins = {}, {}, {} assistant_replays = _AssistantReplays(provider_id, normalize_assistant) async_inputs = _AsyncDeliveryInputs(provider_id, assistant_replays, normalize_user, normalize_full_user) wake_inputs = _MailboxWakeInputs(provider_id, assistant_replays, normalize_user, normalize_full_user) for event, _offset, _line in _records(events, events_stamp): if event.get("session_id") not in (None, "", session_id): continue assistant_replays.event(event) async_inputs.event(event) wake_inputs.event(event) control = _native_control(event, session_id) if control is not None: native_events.append(control) if len(native_events) > MAX_TARGETS: raise _Unproven() run = event.get("run_id") if ( isinstance(run, str) and run and not run.startswith("import_") and event.get("backend") == "claude" and event.get("imported") is not True ): if (event.get("type") == "turn_started" and event.get("purpose") == "scheduled_job" and event.get("job_id") and isinstance(event.get("prompt"), str)): scheduled_starts.setdefault(run, []).append(( _text_key(" ".join(event["prompt"].split())), _timestamp(event.get("ts")), )) elif event.get("type") == "turn_finished": scheduled_ends.setdefault(run, []).append(_timestamp(event.get("ts"))) if len(scheduled_starts) + len(scheduled_ends) > MAX_TARGETS: raise _Unproven() if not isinstance(run, str) or not run.startswith("import_"): continue run_rows[run] = run_rows.get(run, 0) + 1 if len(run_rows) > MAX_TARGETS: raise _Unproven() if event.get("type") == "history_imported": checkpoint = event.get("_history_sync_checkpoint") cursor = checkpoint.get("cursor") if isinstance(checkpoint, dict) else None if ( event.get("backend") == "claude" and event.get("provider_session_id") == provider_id and isinstance(cursor, dict) and cursor.get("version") == 1 and cursor.get("backend") == "claude" and cursor.get("provider_session_id") == provider_id and event.get("source_path") == cursor.get("source_path") and checkpoint.get("version") == 1 and type(event.get("seq")) is int ): if run in batches: raise _Unproven() batches[run] = (event["seq"], checkpoint) elif event.get("type") == "turn_finished" and event.get("imported") is True: if event.get("backend") == "claude" and type(event.get("seq")) is int: terminals[run] = event["seq"] terminal_counts[run] = terminal_counts.get(run, 0) + 1 else: target = _target(event) if target: candidates.append(target) origin = event.get("provider_origin") if isinstance(origin, dict) and origin.get("provider") == "claude": candidate_origins[target] = ( origin.get("event_id"), origin.get("session_id"), origin.get("timestamp"), ) if max(len(batches), len(candidates), len(terminals)) > MAX_TARGETS: raise _Unproven() if not batches or not (candidates or assistant_replays.candidates): return empty paths = {batch[1]["cursor"].get("source_path") for batch in batches.values()} if len(paths) != 1: raise _Unproven() raw_path = paths.pop() if not isinstance(raw_path, str): raise _Unproven() source = Path(raw_path) if not source.is_absolute() or source.suffix != ".jsonl" or source.stem != provider_id: raise _Unproven() if source.is_symlink(): raise _Unproven() source = source.resolve(strict=True) source.relative_to(root.resolve(strict=True)) source_stamp = _stamp(source) prefix_digests = {} eligible_batches = {} for run, (seq, checkpoint) in batches.items(): cursor = checkpoint["cursor"] end = cursor.get("source_offset") start = checkpoint.get("previous_source_offset") expected_digest = cursor.get("source_digest") previous_digest = checkpoint.get("previous_source_digest") if ( type(start) is not int or type(end) is not int or not 0 <= start < end <= source_stamp[2] or (cursor.get("source_dev"), cursor.get("source_ino")) != source_stamp[:2] or not isinstance(expected_digest, str) or not _DIGEST.fullmatch(expected_digest) or type(checkpoint.get("previous_present")) is not bool or (start > 0 and checkpoint.get("previous_present") is not True) or (start == 0 and previous_digest != "") or (start > 0 and (not isinstance(previous_digest, str) or not _DIGEST.fullmatch(previous_digest))) or terminals.get(run, 0) <= seq ): continue prefix_digests.setdefault(end, set()).add(expected_digest) if start: prefix_digests.setdefault(start, set()).add(previous_digest) eligible_batches[run] = (seq, terminals[run], start, end, expected_digest, previous_digest) if not eligible_batches: return _Proof(provider_id, events_stamp, source, source_stamp, frozenset()) digest = hashlib.sha256() verified = set() metadata = {} humans = set() human_offsets = {} interruptions = {} interruption_count = 0 tracker = ClaudeInterruptionTracker() steer_intervals = _steer_intervals(native_events, provider_id) scheduled_ranges = {} for run, starts in scheduled_starts.items(): ends = scheduled_ends.get(run, ()) if len(starts) == len(ends) == 1: key, start_time = starts[0] end_time = ends[0] if start_time is not None and end_time is not None and start_time <= end_time: scheduled_ranges.setdefault(key, []).append((run, start_time, end_time)) scheduled_sources, scheduled_counts = {}, {} for event, offset, line in _records(source, source_stamp): digest.update(line) for wanted in prefix_digests.get(offset, ()): if hmac.compare_digest(digest.hexdigest(), wanted): verified.add((offset, wanted)) assistant_replays.source(event, offset) async_inputs.source(event, offset) wake_inputs.source(event, offset) origin = tracker.consume(event) if event.get("type") != "user": continue text = normalize_user(event) if not isinstance(text, str) or not text: continue key = _text_key(text) if (normalize_full_user is not None and scheduled_ranges and event.get("sessionId") == provider_id and event.get("isMeta") is not True and event.get("isSidechain") is not True): full_text = normalize_full_user(event) source_time = _timestamp(event.get("timestamp")) source_identity = (event.get("uuid"), provider_id, event.get("timestamp")) if isinstance(full_text, str) and isinstance(source_identity[0], str) and source_time is not None: matches = [run for run, start_time, end_time in scheduled_ranges.get( _text_key(" ".join(full_text.split())), (), ) if start_time <= source_time <= end_time] if len(matches) == 1: run = matches[0] scheduled_counts[run] = scheduled_counts.get(run, 0) + 1 scheduled_sources.setdefault((source_identity, key), []).append((offset, run)) origin = _interruption_origin(origin, provider_id, steer_intervals) if origin is not None: interruptions.setdefault(key, []).append((offset, origin)) interruption_count += 1 if interruption_count > MAX_TARGETS: raise _Unproven() if event.get("isMeta") is True or event.get("isCompactSummary") is True or event.get("isSidechain") is True: metadata.setdefault(key, []).append(offset) else: # Preserve the existing global ambiguity rule for isMeta repairs. humans.add(key) if origin is None: human_offsets.setdefault(key, []).append(offset) if len(metadata) + len(humans) + len(interruptions) > MAX_KEYS: raise _Unproven() targets = set() corrected = [] candidate_counts = {} candidate_origin_counts = {} run_candidate_counts = {} for candidate in candidates: _seq, run, key = candidate candidate_counts[(run, key)] = candidate_counts.get((run, key), 0) + 1 origin_key = (run, candidate_origins.get(candidate)) candidate_origin_counts[origin_key] = candidate_origin_counts.get(origin_key, 0) + 1 run_candidate_counts[run] = run_candidate_counts.get(run, 0) + 1 for target in candidates: seq, run, key = target batch = eligible_batches.get(run) if batch is None: continue first_seq, last_seq, start, end, expected, previous = batch if not ( first_seq < seq < last_seq and (end, expected) in verified and (start == 0 or (start, previous) in verified) ): continue # A scheduled wake is ordinary provider user input, not isMeta. Prove # its complete text, native occurrence interval, exact source identity # and original import checkpoint. A shared trimmed prefix is not proof. scheduled = scheduled_sources.get((candidate_origins.get(target), key), ()) if (len(scheduled) == 1 and candidate_origin_counts[(run, candidate_origins.get(target))] == 1 and start < scheduled[0][0] <= end and scheduled_counts.get(scheduled[0][1]) == 1): targets.add(target) continue proven = [(offset, origin) for offset, origin in interruptions.get(key, ()) if start < offset <= end] ordinary = human_offsets.get(key, ()) if ( len(proven) == 1 and candidate_counts[(run, key)] == 1 and bisect_right(ordinary, end) - bisect_right(ordinary, start) == 0 ): corrected.append((target, json.dumps(proven[0][1], sort_keys=True, separators=(",", ":")))) continue # An ambiguous interruption is never demoted to a hidden metadata row. if proven: continue offsets = metadata.get(key, ()) if key not in humans and bisect_right(offsets, end) - bisect_right(offsets, start) == 1: targets.add(target) verified_batches = {run: batch for run, batch in eligible_batches.items() if (batch[3], batch[4]) in verified and (batch[2] == 0 or (batch[2], batch[5]) in verified)} targets.update(async_inputs.prove(verified_batches)) targets.update(wake_inputs.prove(verified_batches)) proven_assistants = assistant_replays.prove(verified_batches) corrected_counts = {} for (_seq, run, _key), _origin in corrected: corrected_counts[run] = corrected_counts.get(run, 0) + 1 for _seq, run, _key in targets: corrected_counts[run] = corrected_counts.get(run, 0) + 1 for _kind, _seq, run, _digest, _identity in proven_assistants: corrected_counts[run] = corrected_counts.get(run, 0) + 1 companions = set() for run, count in corrected_counts.items(): # Only a complete marker-only batch is lifecycle-neutral. Any genuine # user, assistant/tool/unknown row, duplicate terminal or unproven target # leaves both import companions unchanged. if (run_rows[run] == count + 2 and terminal_counts.get(run) == 1): companions.add(("history_imported", eligible_batches[run][0], run)) companions.add(("turn_finished", eligible_batches[run][1], run)) return _Proof(provider_id, events_stamp, source, source_stamp, frozenset(targets), tuple(corrected), frozenset(companions), proven_assistants) def _bounded_records(path: Path, expected, start: int, end: int): """Read complete records in one fixed window, never the oversized prefix.""" if not 0 <= start < end <= expected[2] or end - start > MAX_EVENTS_BYTES: raise _Unproven() descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(descriptor, "rb") as stream: info = os.fstat(stream.fileno()) if (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns) != expected: raise _Unproven() stream.seek(start) if start: stream.seek(start - 1) if stream.read(1) != b"\n": skipped = stream.readline(min(MAX_LINE_BYTES + 1, end - start)) if len(skipped) > MAX_LINE_BYTES or not skipped.endswith(b"\n"): raise _Unproven() count = 0 while stream.tell() < end: offset = stream.tell() line = stream.readline(min(MAX_LINE_BYTES + 1, end - offset)) if len(line) > MAX_LINE_BYTES or not line.endswith(b"\n"): raise _Unproven() count += 1 if count > MAX_RECORDS: raise _Unproven() event = json.loads(line) if not isinstance(event, dict): raise _Unproven() yield event, stream.tell() final = os.fstat(stream.fileno()) if (final.st_dev, final.st_ino, final.st_size, final.st_mtime_ns) != expected: raise _Unproven() if _regular_stamp(path) != expected: raise _Unproven() def _prove_recent_scheduled(session_id: str, provider_id: str, events: Path, root: Path, normalize_user, events_stamp, normalize_full_user, *, normalize_assistant: Callable[[str], str] | None = None, event_window_end: int | None = None) -> _Proof: """Exact-origin scheduled duplicates and provider metadata in recent history. An oversized transcript cannot establish global metadata/interruption proof. Scheduled input requires a complete native occurrence and full source-text match. Metadata requires its explicit structured source flag. Both require an exact, unique origin within the original checkpoint; wording is not proof. Both reads are bounded to 32 MiB; source growth before preparation is allowed. Any change during either read fails visible. """ empty = _Proof(provider_id, events_stamp, None, None, frozenset()) if normalize_full_user is None or not events_stamp[2]: return empty window_end = events_stamp[2] if event_window_end is None else event_window_end if type(window_end) is not int or not 0 < window_end <= events_stamp[2]: raise _Unproven() starts, ends, batches, terminals, rows, candidates = {}, {}, {}, {}, {}, [] origin_counts = {} assistant_replays = _AssistantReplays(provider_id, normalize_assistant) async_inputs = _AsyncDeliveryInputs(provider_id, assistant_replays, normalize_user, normalize_full_user) wake_inputs = _MailboxWakeInputs(provider_id, assistant_replays, normalize_user, normalize_full_user) previous_seq = 0 for event, _offset in _bounded_records( events, events_stamp, max(0, window_end - MAX_EVENTS_BYTES), window_end, ): seq = event.get("seq") if type(seq) is not int or seq <= previous_seq: raise _Unproven() previous_seq = seq if event.get("session_id") not in (None, "", session_id): continue assistant_replays.event(event) async_inputs.event(event) wake_inputs.event(event) run = event.get("run_id") if not isinstance(run, str) or not run: continue if not run.startswith("import_"): if (event.get("backend") != "claude" or event.get("imported") is True or event.get("provider_session_id") not in (None, "", provider_id)): continue if (event.get("type") == "turn_started" and event.get("purpose") == "scheduled_job" and event.get("job_id") and isinstance(event.get("prompt"), str)): starts.setdefault(run, []).append(( _text_key(" ".join(event["prompt"].split())), _timestamp(event.get("ts")), )) elif event.get("type") == "turn_finished": ends.setdefault(run, []).append(_timestamp(event.get("ts"))) else: rows[run] = rows.get(run, 0) + 1 if event.get("type") == "history_imported": checkpoint = event.get("_history_sync_checkpoint") cursor = checkpoint.get("cursor") if isinstance(checkpoint, dict) else None if (event.get("backend") == "claude" and event.get("provider_session_id") == provider_id and isinstance(cursor, dict) and cursor.get("version") == 1 and cursor.get("backend") == "claude" and cursor.get("provider_session_id") == provider_id and checkpoint.get("version") == 1 and event.get("source_path") == cursor.get("source_path")): batches.setdefault(run, []).append((seq, checkpoint)) elif (event.get("type") == "turn_finished" and event.get("imported") is True and event.get("backend") == "claude"): terminals.setdefault(run, []).append(seq) target = _target(event) if target: origin = event.get("provider_origin") identity = (origin.get("event_id"), origin.get("session_id"), origin.get("timestamp")) if isinstance(origin, dict) else () if (isinstance(origin, dict) and origin.get("provider") == "claude" and len(identity) == 3 and all(isinstance(value, str) and value for value in identity) and identity[1] == provider_id and _timestamp(identity[2]) is not None): candidates.append((target, identity)) origin_key = (run, identity) origin_counts[origin_key] = origin_counts.get(origin_key, 0) + 1 if (sum(map(len, (starts, ends, batches, terminals, rows))) > MAX_TARGETS or len(candidates) > MAX_TARGETS): raise _Unproven() if not batches or not (candidates or assistant_replays.candidates): return empty eligible = {} for run, checkpoints in batches.items(): finished = terminals.get(run, ()) if len(checkpoints) != 1 or len(finished) != 1: continue seq, checkpoint = checkpoints[0] cursor = checkpoint["cursor"] start, end = checkpoint.get("previous_source_offset"), cursor.get("source_offset") previous, digest = checkpoint.get("previous_source_digest"), cursor.get("source_digest") if (type(start) is int and type(end) is int and 0 <= start < end and type(checkpoint.get("previous_present")) is bool and ((start == 0 and previous == "") or (start > 0 and checkpoint["previous_present"] is True and isinstance(previous, str) and _DIGEST.fullmatch(previous))) and isinstance(digest, str) and _DIGEST.fullmatch(digest) and seq < finished[0]): eligible[run] = (seq, finished[0], start, end, cursor) if not eligible: return empty paths = {batch[4].get("source_path") for batch in eligible.values()} if len(paths) != 1: raise _Unproven() raw_path = paths.pop() if not isinstance(raw_path, str): raise _Unproven() source = Path(raw_path) if not source.is_absolute() or source.suffix != ".jsonl" or source.stem != provider_id or source.is_symlink(): raise _Unproven() source = source.resolve(strict=True) source.relative_to(root.resolve(strict=True)) source_stamp = _regular_stamp(source) eligible = {run: batch for run, batch in eligible.items() if (batch[4].get("source_dev"), batch[4].get("source_ino")) == source_stamp[:2] and batch[3] <= source_stamp[2]} if not eligible: return empty checkpoint_end = max(batch[3] for batch in eligible.values()) window_start = max(0, checkpoint_end - MAX_EVENTS_BYTES) ranges = {} for run, occurrences in starts.items(): finished = ends.get(run, ()) if len(occurrences) == len(finished) == 1: key, start_time = occurrences[0] end_time = finished[0] if start_time is not None and end_time is not None and start_time <= end_time: ranges.setdefault(key, []).append((run, start_time, end_time)) source_matches, occurrence_counts, identity_counts, source_metadata = {}, {}, {}, {} for event, offset in _bounded_records(source, source_stamp, window_start, checkpoint_end): assistant_replays.source(event, offset) async_inputs.source(event, offset) wake_inputs.source(event, offset) if event.get("type") != "user" or event.get("sessionId") != provider_id: continue identity = (event.get("uuid"), provider_id, event.get("timestamp")) if not all(isinstance(value, str) and value for value in identity): continue identity_counts[identity] = identity_counts.get(identity, 0) + 1 display_text = normalize_user(event) timestamp = _timestamp(identity[2]) if not isinstance(display_text, str) or not display_text or timestamp is None: continue if event.get("isMeta") is True or event.get("isCompactSummary") is True or event.get("isSidechain") is True: source_metadata.setdefault((identity, _text_key(display_text)), []).append(offset) if len(identity_counts) + len(source_metadata) > MAX_KEYS: raise _Unproven() continue full_text = normalize_full_user(event) if not isinstance(full_text, str) or not full_text: continue matches = [run for run, start, end in ranges.get(_text_key(" ".join(full_text.split())), ()) if start <= timestamp <= end] if len(matches) == 1: occurrence = matches[0] occurrence_counts[occurrence] = occurrence_counts.get(occurrence, 0) + 1 source_matches.setdefault((identity, _text_key(display_text)), []).append((offset, occurrence)) if len(identity_counts) + len(source_matches) > MAX_KEYS: raise _Unproven() targets, counts = set(), {} for target, identity in candidates: seq, run, key = target batch, matched = eligible.get(run), source_matches.get((identity, key), ()) if (batch is None or origin_counts[(run, identity)] != 1 or identity_counts.get(identity) != 1): continue first, last, start, end, _cursor = batch metadata = source_metadata.get((identity, key), ()) proven_metadata = len(metadata) == 1 and start < metadata[0] <= end proven_scheduled = (len(matched) == 1 and occurrence_counts.get(matched[0][1]) == 1 and start < matched[0][0] <= end) if first < seq < last and (proven_metadata or proven_scheduled): targets.add(target) counts[run] = counts.get(run, 0) + 1 targets.update(async_inputs.prove(eligible)) targets.update(wake_inputs.prove(eligible)) proven_assistants = assistant_replays.prove(eligible) counts = {} for _seq, run, _key in targets: counts[run] = counts.get(run, 0) + 1 for _kind, _seq, run, _digest, _identity in proven_assistants: counts[run] = counts.get(run, 0) + 1 companions = set() for run, count in counts.items(): if rows[run] == count + 2: companions.add(("history_imported", eligible[run][0], run)) companions.add(("turn_finished", eligible[run][1], run)) return _Proof(provider_id, events_stamp, source, source_stamp, frozenset(targets), companions=frozenset(companions), assistant_replays=proven_assistants) @dataclass(frozen=True) class ClaudeMetadataRepairWindow: """Immutable page-owned proof; lookups never read files or change the main index.""" session_id: str _proof: _Proof = field(repr=False) def is_hidden(self, event: dict) -> bool: if event.get("session_id") not in (None, "", self.session_id): return False try: return _target(event) in self._proof.targets except (ValueError, TypeError): return False def project_event(self, event: dict) -> dict | None: return _project_proof_event(self.session_id, event, self._proof) def _project_proof_event(session_id: str, event: dict, proof: _Proof | None) -> dict | None: if proof is None or event.get("session_id") not in (None, "", session_id): return None if proof.assistant_replays and _assistant_target(event) in proof.assistant_replays: return {**event, "text": "", "metadata_only": True, "provider_history_repair": "source_proven_assistant_replay"} kind, seq, run = event.get("type"), event.get("seq"), event.get("run_id") if (kind in ("history_imported", "turn_finished") and type(seq) is int and isinstance(run, str) and event.get("backend") == "claude" and (kind == "history_imported" or event.get("imported") is True) and (kind, seq, run) in proof.companions and event.get("provider_session_id") in (None, "", proof.provider_id)): return {**event, "imported": True, "metadata_only": True} try: encoded_origin = proof.interruption_index.get(_target(event)) except (ValueError, TypeError): return None if encoded_origin is None: return None origin = json.loads(encoded_origin) projected = dict(event) for key in ("prompt", "text", "_agentsdock_imported_prompt_hidden"): projected.pop(key, None) return {**projected, "type": "provider_interruption", "imported": True, "ts": origin["timestamp"], "provider_origin": origin} class ClaudeMetadataRepairCache: def __init__(self) -> None: self._lock = threading.RLock() # Serialize explicit preparation without blocking per-event lookups # (which can execute on the event loop) behind any filesystem work. self._prepare_lock = threading.Lock() self._proofs: OrderedDict[str, _Proof] = OrderedDict() self._windows: OrderedDict[tuple, ClaudeMetadataRepairWindow] = OrderedDict() self._preparing_session: str | None = None self._preparation_cancelled = False def prepare_window(self, session_id: str, provider_id: str, events: Path, root: Path, normalize_user: Callable[[dict], str | None], *, event_window_end: int, normalize_full_user: Callable[[dict], str | None] | None = None, normalize_assistant: Callable[[str], str] | None = None) -> ClaudeMetadataRepairWindow: """Prove one explicitly requested historical page, never the entire ledger. Unchanged positive and negative windows are cached. Any observed log change, including append growth, requires fresh bounded proof; inode equality alone cannot certify that an old prefix was not rewritten. This cache never contributes to the global timeline-index signature. """ try: stamp = _regular_stamp(events) except (OSError, ValueError): stamp = (0, 0, 0, 0) empty = ClaudeMetadataRepairWindow(session_id, _Proof(provider_id, stamp, None, None, frozenset())) if type(event_window_end) is not int or not 0 < event_window_end <= stamp[2]: return empty key = (session_id, provider_id, str(events.absolute()), str(root.absolute()), stamp, max(0, event_window_end - MAX_EVENTS_BYTES), event_window_end) def cached_window() -> ClaudeMetadataRepairWindow | None: with self._lock: cached = self._windows.get(key) if cached is not None and cached._proof.source is not None: try: if _regular_stamp(cached._proof.source) != cached._proof.source_stamp: return None except (OSError, ValueError): return None with self._lock: if cached is not None and self._windows.get(key) is cached: self._windows.move_to_end(key) return cached return None cached = cached_window() if cached is not None: return cached with self._prepare_lock: cached = cached_window() if cached is not None: return cached with self._lock: self._preparing_session = session_id self._preparation_cancelled = False try: proof = _prove_recent_scheduled(session_id, provider_id, events, root, normalize_user, stamp, normalize_full_user, normalize_assistant=normalize_assistant, event_window_end=event_window_end) window = ClaudeMetadataRepairWindow(session_id, proof) except (OSError, ValueError, TypeError, KeyError, RuntimeError): window = empty with self._lock: cancelled = self._preparation_cancelled self._preparing_session = None if cancelled: return empty self._windows[key] = window self._windows.move_to_end(key) while len(self._windows) > MAX_WINDOWS: self._windows.popitem(last=False) return window def prepare(self, session_id: str, provider_id: str, events: Path, root: Path, normalize_user: Callable[[dict], str | None], *, normalize_full_user: Callable[[dict], str | None] | None = None, normalize_assistant: Callable[[str], str] | None = None, refresh: bool = False) -> bool: """Prepare only this requested session; report a changed suppression map.""" with self._lock: previous = self._proofs.get(session_id) if previous and previous.provider_id == provider_id and not refresh: self._proofs.move_to_end(session_id) return False with self._prepare_lock: with self._lock: previous = self._proofs.get(session_id) if previous and previous.provider_id == provider_id and not refresh: # A verified historical target remains a fact when either # append-only log grows. Do not turn ordinary chat refreshes # into repeated source scans. New imports already filter # metadata; negative admissions are cached as well. self._proofs.move_to_end(session_id) return False self._preparing_session = session_id self._preparation_cancelled = False try: try: stamp = _stamp(events) except _Oversized: if normalize_full_user is None: raise stamp = _regular_stamp(events) # During an explicit import refresh keep the old immutable # proof available to memory-only event readers until replace. if not refresh: with self._lock: self._proofs.pop(session_id, None) try: if stamp[2] > MAX_EVENTS_BYTES: raise _Oversized() proof = _prove(session_id, provider_id, events, root, normalize_user, stamp, normalize_full_user, normalize_assistant) except _Oversized: if normalize_full_user is None: raise proof = _prove_recent_scheduled(session_id, provider_id, events, root, normalize_user, stamp, normalize_full_user, normalize_assistant=normalize_assistant) except (OSError, ValueError, TypeError, KeyError, RuntimeError): # Fail visible, including incomplete or oversized files. A # failed admission must not retry on every page/socket read. proof = _Proof(provider_id, (0, 0, 0, 0), None, None, frozenset()) if (refresh and previous and previous.provider_id == provider_id and previous.source is not None and previous.source == proof.source and previous.source_stamp is not None and proof.source_stamp is not None and previous.events_stamp[:2] == proof.events_stamp[:2] and previous.source_stamp[:2] == proof.source_stamp[:2] and previous.events_stamp[2] <= proof.events_stamp[2] and previous.source_stamp[2] <= proof.source_stamp[2] and (previous.events_stamp[2] < proof.events_stamp[2] or previous.events_stamp == proof.events_stamp) and (previous.source_stamp[2] < proof.source_stamp[2] or previous.source_stamp == proof.source_stamp)): # Exact file ownership and nondecreasing immutable-log prefixes # retain older proofs outside the bounded recent read. A replaced, # truncated, unavailable or different source never inherits them. inputs = previous.targets | proof.targets interruptions = {**dict(previous.interruptions), **dict(proof.interruptions)} assistants = previous.assistant_replays | proof.assistant_replays entries = ([(key[0], key[1], "input", key) for key in inputs] + [(key[0], key[1], "interruption", key) for key in interruptions] + [(key[1], key[2], "assistant", key) for key in assistants]) entries.sort(key=lambda entry: entry[0], reverse=True) evicted_runs = {entry[1] for entry in entries[MAX_TARGETS:]} retained = entries[:MAX_TARGETS] companions = sorted((key for key in previous.companions | proof.companions if key[2] not in evicted_runs), key=lambda key: key[1], reverse=True)[:MAX_TARGETS] # One combined retained-repair budget, not a new allowance on # every import. Evicted proof fails visible, including its # batch companions; no partial batch becomes metadata-only. proof = _Proof(provider_id, proof.events_stamp, proof.source, proof.source_stamp, frozenset(key for _, _, kind, key in retained if kind == "input"), tuple((key, interruptions[key]) for _, _, kind, key in retained if kind == "interruption"), frozenset(companions), frozenset(key for _, _, kind, key in retained if kind == "assistant")) with self._lock: cancelled = self._preparation_cancelled self._preparing_session = None if not cancelled: self._proofs[session_id] = proof self._proofs.move_to_end(session_id) while len(self._proofs) > MAX_SESSIONS: self._proofs.popitem(last=False) else: return bool(previous and previous.signature()) return bool((previous.signature() if previous else frozenset()) != proof.signature()) def signature(self, session_id: str) -> frozenset: """Memory-only cache identity: repaired timeline indexes cannot go stale.""" with self._lock: proof = self._proofs.get(session_id) return proof.signature() if proof else frozenset() def forget(self, session_id: str) -> None: with self._lock: self._proofs.pop(session_id, None) for key in tuple(self._windows): if key[0] == session_id: self._windows.pop(key, None) if self._preparing_session == session_id: self._preparation_cancelled = True def is_hidden(self, session_id: str, event: dict) -> bool: """Memory-only per-event lookup; missing or evicted proof stays visible.""" with self._lock: proof = self._proofs.get(session_id) if not proof or not proof.targets: return False try: target = _target(event) except (ValueError, TypeError): return False return target in proof.targets def interruption_origin(self, session_id: str, event: dict) -> dict | None: """Return copied, verified provenance without filesystem access.""" if event.get("session_id") not in (None, "", session_id): return None with self._lock: proof = self._proofs.get(session_id) if not proof or not proof.interruptions: return None try: target = _target(event) except (ValueError, TypeError): return None origin = proof.interruption_index.get(target) return json.loads(origin) if origin is not None else None def project_interruption(self, session_id: str, event: dict) -> dict | None: origin = self.interruption_origin(session_id, event) if origin is None: return None projected = dict(event) projected.pop("prompt", None) projected.pop("text", None) projected.pop("_agentsdock_imported_prompt_hidden", None) projected.update(type="provider_interruption", imported=True, ts=origin["timestamp"], provider_origin=origin) return projected def project_event(self, session_id: str, event: dict) -> dict | None: """Memory-only correction of exact source-proven historical records.""" with self._lock: proof = self._proofs.get(session_id) return _project_proof_event(session_id, event, proof)