From 312070fa5452b1487d1d1a8c0be8aa907b2d1fa9 Mon Sep 17 00:00:00 2001 From: 8Dionysus Date: Mon, 7 Sep 2026 13:55:14 -0600 Subject: [PATCH 1/3] perf(session-memory): isolate transcript import selection core --- PIPELINE.md | 7 + README.md | 1 + .../validation/validation_evidence_graph.json | 3 + docs/validation/validation_lanes.json | 1 + scripts/AGENTS.md | 2 + scripts/aoa_session_memory.py | 209 +++++------- scripts/aoa_session_memory_import.py | 323 ++++++++++++++++++ scripts/validation_evidence_graph.py | 1 + tests/test_session_memory.py | 39 +++ tests/test_session_memory_import_core.py | 206 +++++++++++ 10 files changed, 677 insertions(+), 115 deletions(-) create mode 100644 scripts/aoa_session_memory_import.py create mode 100644 tests/test_session_memory_import_core.py diff --git a/PIPELINE.md b/PIPELINE.md index 4e74ab67..ac70099a 100644 --- a/PIPELINE.md +++ b/PIPELINE.md @@ -28,6 +28,13 @@ Foreground hooks are bounded and fail-open. A hook failure produces a receipt or incident that later recovery can inspect; it does not make the active agent session depend on archive health. +Historical Codex import uses the small standard-library +`scripts/aoa_session_memory_import.py` core for read-only transcript discovery, +date-window selection, activity-mtime supplements, and optional size-lane +prefiltering. The producer retains the richer title, lineage, and archive +enrichment probe; the core stats current files on each invocation and keeps no +result cache, so a same-size or same-mtime source edit is not silently reused. + The generated Codex command first enters a small standard-library adapter. It atomically persists the exact private hook bytes, byte count, digest, selected roots, event kind, and signal count before returning schema-limited output. diff --git a/README.md b/README.md index 2ab62d0c..db5c168b 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,7 @@ env -u PYTHONDONTWRITEBYTECODE \ PYTHONPYCACHEPREFIX="${PYTHONPYCACHEPREFIX:-${TMPDIR:-/tmp}/aoa-session-memory-pycache}" \ /tmp/aoa-session-memory-venv/bin/python -m pytest -q -p no:cacheprovider \ tests/test_session_memory.py \ + tests/test_session_memory_import_core.py \ tests/test_session_memory_doctor.py \ tests/test_session_memory_outbox.py \ tests/test_session_memory_task_lifecycle.py \ diff --git a/docs/validation/validation_evidence_graph.json b/docs/validation/validation_evidence_graph.json index 252b1718..d1ef52b5 100644 --- a/docs/validation/validation_evidence_graph.json +++ b/docs/validation/validation_evidence_graph.json @@ -101,10 +101,12 @@ "id": "portable-kernel", "patterns": [ "scripts/aoa_session_memory.py", + "scripts/aoa_session_memory_import.py", "scripts/aoa_session_memory_outbox.py", "scripts/aoa_session_memory_privacy.py", "scripts/benchmark_session_projection.py", "tests/test_session_memory.py", + "tests/test_session_memory_import_core.py", "tests/test_session_memory_privacy_core.py", "tests/test_session_memory_outbox_core.py", "tests/session_memory_test_support.py", @@ -223,6 +225,7 @@ "-p", "no:cacheprovider", "tests/test_session_memory.py", + "tests/test_session_memory_import_core.py", "tests/test_session_memory_privacy_core.py", "tests/test_session_memory_outbox_core.py", "tests/test_session_memory_doctor.py", diff --git a/docs/validation/validation_lanes.json b/docs/validation/validation_lanes.json index 9a26e802..d03c865f 100644 --- a/docs/validation/validation_lanes.json +++ b/docs/validation/validation_lanes.json @@ -54,6 +54,7 @@ "-p", "no:cacheprovider", "tests/test_session_memory.py", + "tests/test_session_memory_import_core.py", "tests/test_session_memory_privacy_core.py", "tests/test_session_memory_outbox_core.py", "tests/test_session_memory_doctor.py", diff --git a/scripts/AGENTS.md b/scripts/AGENTS.md index 26b193e0..d04a3700 100644 --- a/scripts/AGENTS.md +++ b/scripts/AGENTS.md @@ -13,6 +13,8 @@ blast radius. - `aoa_session_memory.py` implements archive generation, hook handling, indexing, naming, distillation, validation, export, install, audit, and doctor checks. +- `aoa_session_memory_import.py` owns the bounded, read-only Codex transcript + discovery and date/size selection core used by the producer. - `aoa_epistemic_action_event_chain.py` implements the portable append-only prediction/action/observation chain, replay and concurrency guards, typed discrepancy states, and shadow-only candidate inspection. It is re-exported diff --git a/scripts/aoa_session_memory.py b/scripts/aoa_session_memory.py index ba52920e..440a71f0 100755 --- a/scripts/aoa_session_memory.py +++ b/scripts/aoa_session_memory.py @@ -108,6 +108,62 @@ def _load_outbox_core_module() -> Any: _OUTBOX_CORE = _load_outbox_core_module() + +def _load_import_core_module() -> Any: + """Load the bounded Codex transcript import core from this exact sibling.""" + module_name = "_aoa_session_memory_import_source" + source_digest_attr = "__aoa_session_memory_import_source_sha256__" + source_path = Path(__file__).resolve().with_name( + "aoa_session_memory_import.py" + ) + try: + source_bytes = source_path.read_bytes() + except OSError: + source_bytes = None + source_sha256 = ( + hashlib.sha256(source_bytes).hexdigest() + if source_bytes is not None + else "" + ) + loaded = sys.modules.get(module_name) + loaded_path = getattr(loaded, "__file__", None) + if ( + loaded is not None + and loaded_path is not None + and Path(loaded_path).resolve() == source_path + and source_sha256 + and getattr(loaded, source_digest_attr, None) == source_sha256 + ): + return loaded + spec = importlib.util.spec_from_file_location(module_name, source_path) + if spec is None or spec.loader is None: + raise ImportError("aoa session-memory import core source is unavailable") + module = importlib.util.module_from_spec(spec) + missing = object() + previous = sys.modules.get(module_name, missing) + sys.modules[module_name] = module + try: + if source_bytes is not None and isinstance( + spec.loader, + SourceFileLoader, + ): + # Compile the exact bytes read above so a stale same-path pyc cannot + # satisfy a rapid source edit. + exec( + compile(source_bytes, str(source_path), "exec"), + module.__dict__, + ) + else: + spec.loader.exec_module(module) + setattr(module, source_digest_attr, source_sha256) + except BaseException: + if previous is missing: + sys.modules.pop(module_name, None) + else: + sys.modules[module_name] = previous + raise + return module + PROJECTION_OUTBOX_CONSUMER_MAX_ATTEMPTS = ( _OUTBOX_CORE.PROJECTION_OUTBOX_CONSUMER_MAX_ATTEMPTS ) @@ -283,6 +339,12 @@ def _load_privacy_core_module() -> Any: _EPISTEMIC_ACTION_EVENT_CHAIN = _load_epistemic_action_event_chain_module() _ENTITY_USAGE_PARSERS = _load_entity_usage_parsers_module() _PRIVACY_CORE = _load_privacy_core_module() +_IMPORT_CORE = _load_import_core_module() + +parse_date_arg = _IMPORT_CORE.parse_date_arg +parse_timestamp_arg = _IMPORT_CORE.parse_timestamp_arg +since_date_from_args = _IMPORT_CORE.since_date_from_args +transcript_path_date_hint = _IMPORT_CORE.transcript_path_date_hint for _privacy_name in _PRIVACY_CORE.__all__: globals()[_privacy_name] = getattr(_PRIVACY_CORE, _privacy_name) @@ -868,11 +930,17 @@ def projection_producer_contract_from_source_bytes( "aoa_session_memory_privacy.py" ) ) +SESSION_MEMORY_LOADED_IMPORT_CORE_PATH = ( + SESSION_MEMORY_LOADED_PRODUCER_PATH.with_name( + "aoa_session_memory_import.py" + ) +) SESSION_MEMORY_LOADED_PRODUCER_SOURCE_PATHS = ( SESSION_MEMORY_LOADED_PRODUCER_PATH, SESSION_MEMORY_LOADED_ENTITY_USAGE_PARSER_PATH, SESSION_MEMORY_LOADED_OUTBOX_CORE_PATH, SESSION_MEMORY_LOADED_PRIVACY_CORE_PATH, + SESSION_MEMORY_LOADED_IMPORT_CORE_PATH, ) @@ -901,6 +969,9 @@ def _producer_source_identity_digest( _session_memory_loaded_privacy_core_bytes = ( SESSION_MEMORY_LOADED_PRIVACY_CORE_PATH.read_bytes() ) + _session_memory_loaded_import_core_bytes = ( + SESSION_MEMORY_LOADED_IMPORT_CORE_PATH.read_bytes() + ) except OSError: SESSION_MEMORY_LOADED_PRODUCER_SHA256 = "" else: @@ -930,6 +1001,12 @@ def _producer_source_identity_digest( _session_memory_loaded_privacy_core_bytes ).hexdigest(), ), + ( + SESSION_MEMORY_LOADED_IMPORT_CORE_PATH, + hashlib.sha256( + _session_memory_loaded_import_core_bytes + ).hexdigest(), + ), ) ) SESSION_MEMORY_LOADED_PROJECTION_PRODUCER_CONTRACTS = { @@ -944,6 +1021,7 @@ def _producer_source_identity_digest( _session_memory_loaded_entity_usage_parser_bytes = b"" _session_memory_loaded_outbox_core_bytes = b"" _session_memory_loaded_privacy_core_bytes = b"" + _session_memory_loaded_import_core_bytes = b"" if "SESSION_MEMORY_LOADED_PROJECTION_PRODUCER_CONTRACTS" not in globals(): SESSION_MEMORY_LOADED_PROJECTION_PRODUCER_CONTRACTS: dict[ @@ -18554,60 +18632,6 @@ def transcript_size_prefilter_record( } -def parse_date_arg(value: str | None) -> str | None: - if not value: - return None - match = re.search(r"(20\d{2})[-_]?([01]\d)[-_]?([0-3]\d)", value) - if not match: - raise ValueError(f"expected date like YYYY-MM-DD, got {value!r}") - return f"{match.group(1)}-{match.group(2)}-{match.group(3)}" - - -def parse_timestamp_arg(value: str | None) -> str | None: - if not value: - return None - candidate = str(value).strip() - if re.fullmatch(r"20\d{2}-[01]\d-[0-3]\d", candidate): - candidate = f"{candidate}T00:00:00Z" - normalized = candidate[:-1] + "+00:00" if candidate.endswith("Z") else candidate - try: - parsed = datetime.fromisoformat(normalized) - except ValueError as exc: - raise ValueError(f"expected ISO-8601 timestamp, got {value!r}") from exc - if parsed.tzinfo is None: - parsed = parsed.replace(tzinfo=timezone.utc) - return parsed.astimezone(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z") - - -def since_date_from_args(since: str | None, since_days: int | None) -> str | None: - explicit = parse_date_arg(since) - if explicit: - return explicit - if since_days is None: - return None - return (datetime.now(timezone.utc) - timedelta(days=since_days)).strftime("%Y-%m-%d") - - -def transcript_path_date_hint(raw_path: Path, source_root: Path) -> str | None: - try: - parts = raw_path.relative_to(source_root).parts - except ValueError: - parts = raw_path.parts - for index in range(max(0, len(parts) - 2)): - year, month, day = parts[index : index + 3] - if not (re.fullmatch(r"20\d{2}", year) and re.fullmatch(r"[01]\d", month) and re.fullmatch(r"[0-3]\d", day)): - continue - try: - datetime(int(year), int(month), int(day), tzinfo=timezone.utc) - except ValueError: - continue - return f"{year}-{month}-{day}" - try: - return parse_date_arg(raw_path.name) - except ValueError: - return None - - def discover_codex_transcripts( *, source_root: Path, @@ -18617,71 +18641,24 @@ def discover_codex_transcripts( min_raw_bytes: int | None = None, max_raw_bytes: int | None = None, ) -> list[dict[str, Any]]: - source_root = source_root.expanduser() - if not source_root.exists(): - return [] - since_date = parse_date_arg(since) - until_date = parse_date_arg(until) - records: list[dict[str, Any]] = [] - for raw_path in sorted(source_root.rglob("*.jsonl")): - if not raw_path.is_file(): - continue - try: - source_stat = raw_path.stat() - except OSError: - continue - activity_window_match = bool( - activity_since_epoch is not None - and source_stat.st_mtime >= float(activity_since_epoch) - ) - path_date = transcript_path_date_hint(raw_path, source_root) - path_date_outside_window = False - if path_date: - if since_date and path_date < since_date: - path_date_outside_window = True - if until_date and path_date > until_date: - path_date_outside_window = True - if path_date_outside_window and not activity_window_match: - continue - outside_size_lane = bool( - ( - min_raw_bytes is not None - and source_stat.st_size < int(min_raw_bytes) - ) - or ( - max_raw_bytes is not None - and source_stat.st_size > int(max_raw_bytes) - ) - ) - if outside_size_lane: - record = transcript_size_prefilter_record( + return _IMPORT_CORE.discover_codex_transcripts( + source_root=source_root, + since=since, + until=until, + activity_since_epoch=activity_since_epoch, + min_raw_bytes=min_raw_bytes, + max_raw_bytes=max_raw_bytes, + # Keep the producer's richer metadata/title/lineage probe while the + # bounded core owns traversal, date windows, and stat-based selection. + transcript_probe=transcript_probe, + transcript_size_prefilter_record=( + lambda raw_path, source_stat, path_date: transcript_size_prefilter_record( raw_path, source_stat=source_stat, path_date=path_date, ) - else: - record = transcript_probe(raw_path) - session_date = str(record.get("session_date") or "") - session_date_outside_window = bool( - (since_date and session_date < since_date) - or (until_date and session_date > until_date) - ) - if session_date_outside_window and not activity_window_match: - continue - activity_supplement = bool( - activity_window_match - and (path_date_outside_window or session_date_outside_window) - ) - record["source_mtime_epoch"] = source_stat.st_mtime - record["selection_source"] = ( - "activity_mtime_supplement" - if activity_supplement - else "date_window" - ) - record["activity_window_match"] = activity_window_match - records.append(record) - records.sort(key=lambda item: (str(item.get("session_date") or ""), str(item.get("timestamp") or ""), str(item.get("transcript_path") or ""))) - return records + ), + ) def existing_archive_by_session_id(aoa_root: Path) -> dict[str, dict[str, Any]]: @@ -224784,6 +224761,7 @@ def command_audit(args: argparse.Namespace) -> int: REQUIRED_TEST_ROOT_FILES = [ "tests/AGENTS.md", "tests/test_session_memory.py", + "tests/test_session_memory_import_core.py", "tests/session_memory_test_support.py", "tests/test_session_memory_doctor.py", "tests/test_session_memory_outbox.py", @@ -224893,6 +224871,7 @@ def command_audit(args: argparse.Namespace) -> int: "scripts/aoa_session_memory_entity_usage_parsers.py", "scripts/aoa_session_memory_outbox.py", "scripts/aoa_session_memory_privacy.py", + "scripts/aoa_session_memory_import.py", "scripts/validate_local_stats_port.py", "sessions/AGENTS.md", "skills/AGENTS.md", diff --git a/scripts/aoa_session_memory_import.py b/scripts/aoa_session_memory_import.py new file mode 100644 index 00000000..2d5de4dd --- /dev/null +++ b/scripts/aoa_session_memory_import.py @@ -0,0 +1,323 @@ +"""Small, source-only boundary for Codex transcript candidate collection. + +The session-memory producer owns enrichment and archive synchronization. This +module owns only the bounded, read-only part of historical import selection so +that collection policy can be tested without importing the full producer. +Callbacks let the producer keep its richer title/lineage probe while the +standalone route remains useful with its conservative metadata probe. +""" + +from __future__ import annotations + +import json +import os +import re +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable + + +TranscriptProbe = Callable[[Path], dict[str, Any]] +SizePrefilter = Callable[ + [Path, os.stat_result, str | None], + dict[str, Any], +] + + +def parse_date_arg(value: str | None) -> str | None: + if not value: + return None + match = re.search(r"(20\d{2})[-_]?([01]\d)[-_]?([0-3]\d)", value) + if not match: + raise ValueError(f"expected date like YYYY-MM-DD, got {value!r}") + return f"{match.group(1)}-{match.group(2)}-{match.group(3)}" + + +def parse_timestamp_arg(value: str | None) -> str | None: + if not value: + return None + candidate = str(value).strip() + if re.fullmatch(r"20\d{2}-[01]\d-[0-3]\d", candidate): + candidate = f"{candidate}T00:00:00Z" + normalized = candidate[:-1] + "+00:00" if candidate.endswith("Z") else candidate + try: + parsed = datetime.fromisoformat(normalized) + except ValueError as exc: + raise ValueError(f"expected ISO-8601 timestamp, got {value!r}") from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).isoformat(timespec="microseconds").replace( + "+00:00", "Z" + ) + + +def since_date_from_args(since: str | None, since_days: int | None) -> str | None: + explicit = parse_date_arg(since) + if explicit: + return explicit + if since_days is None: + return None + return (datetime.now(timezone.utc) - timedelta(days=since_days)).strftime( + "%Y-%m-%d" + ) + + +def transcript_path_date_hint(raw_path: Path, source_root: Path) -> str | None: + """Read a YYYY/MM/DD path hint without opening the transcript.""" + + try: + parts = raw_path.relative_to(source_root).parts + except ValueError: + parts = raw_path.parts + for index in range(max(0, len(parts) - 2)): + year, month, day = parts[index : index + 3] + if not ( + re.fullmatch(r"20\d{2}", year) + and re.fullmatch(r"[01]\d", month) + and re.fullmatch(r"[0-3]\d", day) + ): + continue + try: + datetime(int(year), int(month), int(day), tzinfo=timezone.utc) + except ValueError: + continue + return f"{year}-{month}-{day}" + try: + return parse_date_arg(raw_path.name) + except ValueError: + return None + + +def _default_metadata_probe( + raw_path: Path, + *, + source_root: Path | None = None, +) -> dict[str, Any]: + """Conservatively identify a transcript without importing the producer.""" + + event: dict[str, Any] = {"transcript_path": str(raw_path)} + try: + with raw_path.open("r", encoding="utf-8", errors="replace") as handle: + for line_no, line in enumerate(handle, start=1): + if line_no > 40: + break + try: + parsed = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(parsed, dict) or parsed.get("type") != "session_meta": + continue + payload = parsed.get("payload") + if not isinstance(payload, dict): + continue + for key in ( + "id", + "cwd", + "timestamp", + "model", + "model_provider", + "cli_version", + ): + if payload.get(key): + event[key] = payload[key] + break + except OSError: + pass + source_stat = raw_path.stat() + path_date = transcript_path_date_hint( + raw_path, + source_root if source_root is not None else raw_path.parent, + ) + timestamp = str(event.get("timestamp") or "") + try: + session_date = parse_date_arg(timestamp) + except ValueError: + session_date = None + session_id = str(event.get("id") or raw_path.stem) + return { + "session_id": session_id, + "transcript_path": str(raw_path), + "session_date": session_date or path_date or datetime.fromtimestamp( + source_stat.st_mtime, timezone.utc + ).strftime("%Y-%m-%d"), + "title": raw_path.stem, + "title_source": "transcript_path_metadata_probe", + "cwd": event.get("cwd"), + "timestamp": event.get("timestamp"), + "model": event.get("model"), + "model_provider": event.get("model_provider"), + "cli_version": event.get("cli_version"), + "lineage": {}, + "bytes": source_stat.st_size, + "mtime": datetime.fromtimestamp( + source_stat.st_mtime, timezone.utc + ).strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +def _default_size_prefilter( + raw_path: Path, + source_stat: os.stat_result, + path_date: str | None, +) -> dict[str, Any]: + """Read only the identity prefix for a transcript outside the size lane.""" + + event: dict[str, Any] = {"transcript_path": str(raw_path)} + try: + with raw_path.open("r", encoding="utf-8", errors="replace") as handle: + for line_no, line in enumerate(handle, start=1): + if line_no > 40: + break + try: + parsed = json.loads(line) + except json.JSONDecodeError: + continue + payload = ( + parsed.get("payload") + if isinstance(parsed, dict) + and isinstance(parsed.get("payload"), dict) + else {} + ) + if not isinstance(parsed, dict) or parsed.get("type") != "session_meta": + continue + if payload.get("id"): + event["id"] = payload["id"] + for key in ( + "cwd", + "timestamp", + "model", + "model_provider", + "cli_version", + ): + if payload.get(key): + event[key] = payload[key] + break + except OSError: + pass + try: + timestamp_date = parse_date_arg(str(event.get("timestamp") or "")) + except ValueError: + timestamp_date = None + return { + "session_id": str(event.get("id") or raw_path.stem), + "transcript_path": str(raw_path), + "session_date": timestamp_date or path_date or datetime.fromtimestamp( + source_stat.st_mtime, timezone.utc + ).strftime("%Y-%m-%d"), + "title": raw_path.stem, + "title_source": "transcript_path_size_prefilter", + "cwd": event.get("cwd"), + "timestamp": event.get("timestamp"), + "model": event.get("model"), + "model_provider": event.get("model_provider"), + "cli_version": event.get("cli_version"), + "lineage": {}, + "bytes": source_stat.st_size, + "mtime": datetime.fromtimestamp( + source_stat.st_mtime, timezone.utc + ).strftime("%Y-%m-%dT%H:%M:%SZ"), + "metadata_probe_status": "identity_only_size_prefilter", + } + + +def discover_codex_transcripts( + *, + source_root: Path, + since: str | None = None, + until: str | None = None, + activity_since_epoch: float | None = None, + min_raw_bytes: int | None = None, + max_raw_bytes: int | None = None, + transcript_probe: TranscriptProbe | None = None, + transcript_size_prefilter_record: SizePrefilter | None = None, +) -> list[dict[str, Any]]: + """Collect candidate transcripts using only bounded source metadata. + + No result cache is consulted. Every invocation stats the current files, + and the supplied callbacks are called after that stat, preserving the + producer's richer source/title/lineage behavior when it is available. + """ + + source_root = source_root.expanduser() + if not source_root.exists(): + return [] + since_date = parse_date_arg(since) + until_date = parse_date_arg(until) + probe = transcript_probe or ( + lambda raw_path: _default_metadata_probe( + raw_path, + source_root=source_root, + ) + ) + size_prefilter = transcript_size_prefilter_record or _default_size_prefilter + records: list[dict[str, Any]] = [] + for raw_path in sorted(source_root.rglob("*.jsonl")): + if not raw_path.is_file(): + continue + try: + source_stat = raw_path.stat() + except OSError: + continue + activity_window_match = bool( + activity_since_epoch is not None + and source_stat.st_mtime >= float(activity_since_epoch) + ) + path_date = transcript_path_date_hint(raw_path, source_root) + path_date_outside_window = False + if path_date: + if since_date and path_date < since_date: + path_date_outside_window = True + if until_date and path_date > until_date: + path_date_outside_window = True + if path_date_outside_window and not activity_window_match: + continue + outside_size_lane = bool( + ( + min_raw_bytes is not None + and source_stat.st_size < int(min_raw_bytes) + ) + or ( + max_raw_bytes is not None + and source_stat.st_size > int(max_raw_bytes) + ) + ) + if outside_size_lane: + record = size_prefilter(raw_path, source_stat, path_date) + else: + record = probe(raw_path) + session_date = str(record.get("session_date") or "") + session_date_outside_window = bool( + (since_date and session_date < since_date) + or (until_date and session_date > until_date) + ) + if session_date_outside_window and not activity_window_match: + continue + activity_supplement = bool( + activity_window_match + and (path_date_outside_window or session_date_outside_window) + ) + record["source_mtime_epoch"] = source_stat.st_mtime + record["selection_source"] = ( + "activity_mtime_supplement" + if activity_supplement + else "date_window" + ) + record["activity_window_match"] = activity_window_match + records.append(record) + records.sort( + key=lambda item: ( + str(item.get("session_date") or ""), + str(item.get("timestamp") or ""), + str(item.get("transcript_path") or ""), + ) + ) + return records + + +__all__ = [ + "discover_codex_transcripts", + "parse_date_arg", + "parse_timestamp_arg", + "since_date_from_args", + "transcript_path_date_hint", +] diff --git a/scripts/validation_evidence_graph.py b/scripts/validation_evidence_graph.py index 261596a4..cd1bf194 100644 --- a/scripts/validation_evidence_graph.py +++ b/scripts/validation_evidence_graph.py @@ -28,6 +28,7 @@ "-p", "no:cacheprovider", "tests/test_session_memory.py", + "tests/test_session_memory_import_core.py", "tests/test_session_memory_privacy_core.py", "tests/test_session_memory_outbox_core.py", "tests/test_session_memory_doctor.py", diff --git a/tests/test_session_memory.py b/tests/test_session_memory.py index 640e8656..f6099f3b 100644 --- a/tests/test_session_memory.py +++ b/tests/test_session_memory.py @@ -66597,6 +66597,45 @@ def test_outbox_core_loader_binds_exact_sibling_over_foreign_cached_module( assert second is not first +def test_import_core_loader_binds_exact_sibling_over_foreign_cached_module( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + private_name = "_aoa_session_memory_import_source" + fixture_main = tmp_path / "aoa_session_memory.py" + fixture_core = tmp_path / "aoa_session_memory_import.py" + fixture_main.write_text("# fixture runtime\n", encoding="utf-8") + fixture_core.write_bytes(b'VALUE = "v1"\n') + initial_mtime_ns = fixture_core.stat().st_mtime_ns + foreign_path = tmp_path / "foreign" / "aoa_session_memory_import.py" + foreign_path.parent.mkdir() + foreign_path.write_bytes(b'VALUE = "foreign"\n') + foreign_spec = importlib.util.spec_from_file_location( + private_name, + foreign_path, + ) + assert foreign_spec is not None and foreign_spec.loader is not None + foreign = importlib.util.module_from_spec(foreign_spec) + foreign_spec.loader.exec_module(foreign) + foreign.VALUE = "foreign" + monkeypatch.setattr(module, "__file__", str(fixture_main)) + monkeypatch.setitem(sys.modules, private_name, foreign) + + first = module._load_import_core_module() + assert first is not foreign + assert first.VALUE == "v1" + assert Path(first.__file__).resolve() == fixture_core.resolve() + assert module._load_import_core_module() is first + + # Keep size and mtime stable so a timestamp-based bytecode cache cannot + # explain the source-edit reload. + fixture_core.write_bytes(b'VALUE = "v2"\n') + os.utime(fixture_core, ns=(initial_mtime_ns, initial_mtime_ns)) + second = module._load_import_core_module() + assert second.VALUE == "v2" + assert second is not first + + def test_stage_work_identity_isolates_session_index_change( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_session_memory_import_core.py b/tests/test_session_memory_import_core.py new file mode 100644 index 00000000..970f442a --- /dev/null +++ b/tests/test_session_memory_import_core.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import importlib.util +import os +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +IMPORT_CORE = ROOT / "scripts" / "aoa_session_memory_import.py" + + +def load_import_core() -> ModuleType: + spec = importlib.util.spec_from_file_location( + "aoa_session_memory_import_test_source", + IMPORT_CORE, + ) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +module = load_import_core() + + +def write_transcript( + path: Path, + *, + session_id: str, + timestamp: str, + body: str = "event", +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + "\n".join( + [ + '{"type":"session_meta","payload":' + f'{{"id":"{session_id}","timestamp":"{timestamp}",' + '"cwd":"/tmp/workspace"}}', + f'{{"type":"event","body":"{body}"}}', + ] + ) + + "\n", + encoding="utf-8", + ) + + +def epoch(value: str) -> float: + return datetime.fromisoformat(value).replace(tzinfo=timezone.utc).timestamp() + + +def test_date_helpers_keep_bounded_normalization_and_reject_invalid_values() -> None: + assert module.parse_date_arg("rollout_2026-09-07.jsonl") == "2026-09-07" + assert module.parse_date_arg("20260907") == "2026-09-07" + assert module.parse_timestamp_arg("2026-09-07") == "2026-09-07T00:00:00.000000Z" + assert module.parse_timestamp_arg("2026-09-07T03:04:05-02:00") == ( + "2026-09-07T05:04:05.000000Z" + ) + with pytest.raises(ValueError): + module.parse_date_arg("not-a-date") + with pytest.raises(ValueError): + module.parse_timestamp_arg("not-a-timestamp") + + +def test_path_date_hint_uses_nested_codex_date_directories(tmp_path: Path) -> None: + source_root = tmp_path / "sessions" + transcript = source_root / "2026" / "09" / "07" / "rollout.jsonl" + assert module.transcript_path_date_hint(transcript, source_root) == "2026-09-07" + named = source_root / "rollout-2026_09_08.jsonl" + assert module.transcript_path_date_hint(named, source_root) == "2026-09-08" + invalid = source_root / "2026" / "99" / "99" / "rollout.jsonl" + assert module.transcript_path_date_hint(invalid, source_root) is None + + +def test_discovery_applies_date_and_activity_windows_with_deterministic_order( + tmp_path: Path, +) -> None: + source_root = tmp_path / "sessions" + old = source_root / "2026" / "09" / "01" / "old.jsonl" + in_window = source_root / "2026" / "09" / "07" / "in-window.jsonl" + active_old = source_root / "2026" / "08" / "31" / "active-old.jsonl" + write_transcript(old, session_id="old", timestamp="2026-09-01T00:00:00Z") + write_transcript( + in_window, + session_id="in-window", + timestamp="2026-09-07T00:00:00Z", + ) + write_transcript( + active_old, + session_id="active-old", + timestamp="2026-08-31T00:00:00Z", + ) + old_mtime = epoch("2026-09-01T00:00:00") + active_mtime = epoch("2026-09-09T00:00:00") + os.utime(old, (old_mtime, old_mtime)) + os.utime(in_window, (old_mtime, old_mtime)) + os.utime(active_old, (active_mtime, active_mtime)) + + records = module.discover_codex_transcripts( + source_root=source_root, + since="2026-09-07", + activity_since_epoch=epoch("2026-09-08T00:00:00"), + ) + + assert [item["session_id"] for item in records] == ["active-old", "in-window"] + assert records[0]["selection_source"] == "activity_mtime_supplement" + assert records[0]["activity_window_match"] is True + assert records[1]["selection_source"] == "date_window" + assert records[1]["activity_window_match"] is False + + +def test_discovery_size_lane_uses_identity_prefilter_without_full_probe( + tmp_path: Path, +) -> None: + source_root = tmp_path / "sessions" + small = source_root / "2026" / "09" / "07" / "small.jsonl" + normal = source_root / "2026" / "09" / "07" / "normal.jsonl" + write_transcript(small, session_id="small", timestamp="2026-09-07T00:00:00Z") + write_transcript(normal, session_id="normal", timestamp="2026-09-07T01:00:00Z") + probe_calls: list[str] = [] + prefilter_calls: list[str] = [] + + def probe(path: Path) -> dict[str, object]: + probe_calls.append(path.name) + return { + "session_id": path.stem, + "transcript_path": str(path), + "session_date": "2026-09-07", + "timestamp": "2026-09-07T01:00:00Z", + } + + def prefilter( + path: Path, + stat_result: os.stat_result, + path_date: str | None, + ) -> dict[str, object]: + prefilter_calls.append(path.name) + return { + "session_id": path.stem, + "transcript_path": str(path), + "session_date": path_date, + "timestamp": None, + } + + records = module.discover_codex_transcripts( + source_root=source_root, + min_raw_bytes=normal.stat().st_size + 1, + transcript_probe=probe, + transcript_size_prefilter_record=prefilter, + ) + + assert probe_calls == [] + assert prefilter_calls == ["normal.jsonl", "small.jsonl"] + assert [item["session_id"] for item in records] == ["normal", "small"] + + +def test_discovery_reads_current_bytes_on_same_size_and_mtime_mutation( + tmp_path: Path, +) -> None: + source_root = tmp_path / "sessions" + transcript = source_root / "2026" / "09" / "07" / "rollout.jsonl" + write_transcript( + transcript, + session_id="alpha", + timestamp="2026-09-07T00:00:00Z", + body="alpha", + ) + original_stat = transcript.stat() + seen: list[str] = [] + + def probe(path: Path) -> dict[str, object]: + body = path.read_text(encoding="utf-8") + seen.append(body) + return { + "session_id": "alpha" if "alpha" in body else "bravo", + "transcript_path": str(path), + "session_date": "2026-09-07", + "timestamp": "2026-09-07T00:00:00Z", + } + + first = module.discover_codex_transcripts( + source_root=source_root, + transcript_probe=probe, + ) + assert first[0]["session_id"] == "alpha" + + write_transcript( + transcript, + session_id="bravo", + timestamp="2026-09-07T00:00:00Z", + body="bravo", + ) + assert transcript.stat().st_size == original_stat.st_size + os.utime(transcript, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns)) + + second = module.discover_codex_transcripts( + source_root=source_root, + transcript_probe=probe, + ) + assert second[0]["session_id"] == "bravo" + assert len(seen) == 2 + assert "bravo" in seen[1] From f912fdf94c289a49fcb57fa19e8a543ad15a01e0 Mon Sep 17 00:00:00 2001 From: 8Dionysus Date: Mon, 7 Sep 2026 14:10:41 -0600 Subject: [PATCH 2/3] refactor(session-memory): keep import parser in producer --- scripts/aoa_session_memory_import.py | 161 ++--------------------- tests/test_session_memory_import_core.py | 20 +++ 2 files changed, 31 insertions(+), 150 deletions(-) diff --git a/scripts/aoa_session_memory_import.py b/scripts/aoa_session_memory_import.py index 2d5de4dd..b077f0b0 100644 --- a/scripts/aoa_session_memory_import.py +++ b/scripts/aoa_session_memory_import.py @@ -1,15 +1,14 @@ -"""Small, source-only boundary for Codex transcript candidate collection. +"""Small, read-only boundary for Codex transcript candidate selection. The session-memory producer owns enrichment and archive synchronization. This -module owns only the bounded, read-only part of historical import selection so +module owns only the read-only part of historical import selection so that collection policy can be tested without importing the full producer. -Callbacks let the producer keep its richer title/lineage probe while the -standalone route remains useful with its conservative metadata probe. +Required callbacks keep metadata, title, and lineage parsing with the producer; +this module does not introduce a second transcript parser. """ from __future__ import annotations -import json import os import re from datetime import datetime, timedelta, timezone @@ -88,154 +87,23 @@ def transcript_path_date_hint(raw_path: Path, source_root: Path) -> str | None: return None -def _default_metadata_probe( - raw_path: Path, - *, - source_root: Path | None = None, -) -> dict[str, Any]: - """Conservatively identify a transcript without importing the producer.""" - - event: dict[str, Any] = {"transcript_path": str(raw_path)} - try: - with raw_path.open("r", encoding="utf-8", errors="replace") as handle: - for line_no, line in enumerate(handle, start=1): - if line_no > 40: - break - try: - parsed = json.loads(line) - except json.JSONDecodeError: - continue - if not isinstance(parsed, dict) or parsed.get("type") != "session_meta": - continue - payload = parsed.get("payload") - if not isinstance(payload, dict): - continue - for key in ( - "id", - "cwd", - "timestamp", - "model", - "model_provider", - "cli_version", - ): - if payload.get(key): - event[key] = payload[key] - break - except OSError: - pass - source_stat = raw_path.stat() - path_date = transcript_path_date_hint( - raw_path, - source_root if source_root is not None else raw_path.parent, - ) - timestamp = str(event.get("timestamp") or "") - try: - session_date = parse_date_arg(timestamp) - except ValueError: - session_date = None - session_id = str(event.get("id") or raw_path.stem) - return { - "session_id": session_id, - "transcript_path": str(raw_path), - "session_date": session_date or path_date or datetime.fromtimestamp( - source_stat.st_mtime, timezone.utc - ).strftime("%Y-%m-%d"), - "title": raw_path.stem, - "title_source": "transcript_path_metadata_probe", - "cwd": event.get("cwd"), - "timestamp": event.get("timestamp"), - "model": event.get("model"), - "model_provider": event.get("model_provider"), - "cli_version": event.get("cli_version"), - "lineage": {}, - "bytes": source_stat.st_size, - "mtime": datetime.fromtimestamp( - source_stat.st_mtime, timezone.utc - ).strftime("%Y-%m-%dT%H:%M:%SZ"), - } - - -def _default_size_prefilter( - raw_path: Path, - source_stat: os.stat_result, - path_date: str | None, -) -> dict[str, Any]: - """Read only the identity prefix for a transcript outside the size lane.""" - - event: dict[str, Any] = {"transcript_path": str(raw_path)} - try: - with raw_path.open("r", encoding="utf-8", errors="replace") as handle: - for line_no, line in enumerate(handle, start=1): - if line_no > 40: - break - try: - parsed = json.loads(line) - except json.JSONDecodeError: - continue - payload = ( - parsed.get("payload") - if isinstance(parsed, dict) - and isinstance(parsed.get("payload"), dict) - else {} - ) - if not isinstance(parsed, dict) or parsed.get("type") != "session_meta": - continue - if payload.get("id"): - event["id"] = payload["id"] - for key in ( - "cwd", - "timestamp", - "model", - "model_provider", - "cli_version", - ): - if payload.get(key): - event[key] = payload[key] - break - except OSError: - pass - try: - timestamp_date = parse_date_arg(str(event.get("timestamp") or "")) - except ValueError: - timestamp_date = None - return { - "session_id": str(event.get("id") or raw_path.stem), - "transcript_path": str(raw_path), - "session_date": timestamp_date or path_date or datetime.fromtimestamp( - source_stat.st_mtime, timezone.utc - ).strftime("%Y-%m-%d"), - "title": raw_path.stem, - "title_source": "transcript_path_size_prefilter", - "cwd": event.get("cwd"), - "timestamp": event.get("timestamp"), - "model": event.get("model"), - "model_provider": event.get("model_provider"), - "cli_version": event.get("cli_version"), - "lineage": {}, - "bytes": source_stat.st_size, - "mtime": datetime.fromtimestamp( - source_stat.st_mtime, timezone.utc - ).strftime("%Y-%m-%dT%H:%M:%SZ"), - "metadata_probe_status": "identity_only_size_prefilter", - } - - def discover_codex_transcripts( *, source_root: Path, + transcript_probe: TranscriptProbe, + transcript_size_prefilter_record: SizePrefilter, since: str | None = None, until: str | None = None, activity_since_epoch: float | None = None, min_raw_bytes: int | None = None, max_raw_bytes: int | None = None, - transcript_probe: TranscriptProbe | None = None, - transcript_size_prefilter_record: SizePrefilter | None = None, ) -> list[dict[str, Any]]: - """Collect candidate transcripts using only bounded source metadata. + """Select candidate transcripts from current paths, stats, and probe records. No result cache is consulted. Every invocation stats the current files, and the supplied callbacks are called after that stat, preserving the - producer's richer source/title/lineage behavior when it is available. + producer's source/title/lineage behavior. Discovery walks the complete + source tree; this extraction adds no file-count or transcript-byte budget. """ source_root = source_root.expanduser() @@ -243,13 +111,6 @@ def discover_codex_transcripts( return [] since_date = parse_date_arg(since) until_date = parse_date_arg(until) - probe = transcript_probe or ( - lambda raw_path: _default_metadata_probe( - raw_path, - source_root=source_root, - ) - ) - size_prefilter = transcript_size_prefilter_record or _default_size_prefilter records: list[dict[str, Any]] = [] for raw_path in sorted(source_root.rglob("*.jsonl")): if not raw_path.is_file(): @@ -282,9 +143,9 @@ def discover_codex_transcripts( ) ) if outside_size_lane: - record = size_prefilter(raw_path, source_stat, path_date) + record = transcript_size_prefilter_record(raw_path, source_stat, path_date) else: - record = probe(raw_path) + record = transcript_probe(raw_path) session_date = str(record.get("session_date") or "") session_date_outside_window = bool( (since_date and session_date < since_date) diff --git a/tests/test_session_memory_import_core.py b/tests/test_session_memory_import_core.py index 970f442a..2a04f812 100644 --- a/tests/test_session_memory_import_core.py +++ b/tests/test_session_memory_import_core.py @@ -1,6 +1,7 @@ from __future__ import annotations import importlib.util +import json import os from datetime import datetime, timezone from pathlib import Path @@ -53,6 +54,21 @@ def epoch(value: str) -> float: return datetime.fromisoformat(value).replace(tzinfo=timezone.utc).timestamp() +def candidate_record(path: Path) -> dict[str, object]: + """Records for these synthetic fixtures, not another production parser.""" + payload = json.loads(path.read_text().splitlines()[0])["payload"] + return { + "session_id": payload["id"], + "transcript_path": str(path), + "session_date": payload["timestamp"][:10], + "timestamp": payload["timestamp"], + } + + +def unexpected_size_prefilter(*_args: object) -> dict[str, object]: + pytest.fail("a transcript inside the size lane must use the producer probe") + + def test_date_helpers_keep_bounded_normalization_and_reject_invalid_values() -> None: assert module.parse_date_arg("rollout_2026-09-07.jsonl") == "2026-09-07" assert module.parse_date_arg("20260907") == "2026-09-07" @@ -102,6 +118,8 @@ def test_discovery_applies_date_and_activity_windows_with_deterministic_order( records = module.discover_codex_transcripts( source_root=source_root, + transcript_probe=candidate_record, + transcript_size_prefilter_record=unexpected_size_prefilter, since="2026-09-07", activity_since_epoch=epoch("2026-09-08T00:00:00"), ) @@ -185,6 +203,7 @@ def probe(path: Path) -> dict[str, object]: first = module.discover_codex_transcripts( source_root=source_root, transcript_probe=probe, + transcript_size_prefilter_record=unexpected_size_prefilter, ) assert first[0]["session_id"] == "alpha" @@ -200,6 +219,7 @@ def probe(path: Path) -> dict[str, object]: second = module.discover_codex_transcripts( source_root=source_root, transcript_probe=probe, + transcript_size_prefilter_record=unexpected_size_prefilter, ) assert second[0]["session_id"] == "bravo" assert len(seen) == 2 From 096e83224ff4aa529186e5acea2c01571e47911a Mon Sep 17 00:00:00 2001 From: 8Dionysus Date: Mon, 7 Sep 2026 14:34:47 -0600 Subject: [PATCH 3/3] docs(session-memory): document import core feedback route --- VALIDATION.md | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/VALIDATION.md b/VALIDATION.md index 96e263b7..70da6b24 100644 --- a/VALIDATION.md +++ b/VALIDATION.md @@ -39,11 +39,17 @@ outbox_core_pycache="$(mktemp -d "${TMPDIR:-/tmp}/aoa-session-memory-outbox.XXXX env -u PYTHONDONTWRITEBYTECODE PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 PYTHONPYCACHEPREFIX="$outbox_core_pycache" \ python3 -m pytest -q -p no:cacheprovider --rootdir=. --confcutdir=. \ tests/test_session_memory_outbox_core.py +# Transcript-import sibling edit: +import_core_pycache="$(mktemp -d "${TMPDIR:-/tmp}/aoa-session-memory-import.XXXXXX")" +env -u PYTHONDONTWRITEBYTECODE PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 PYTHONPYCACHEPREFIX="$import_core_pycache" \ + python3 -m pytest -q -p no:cacheprovider --rootdir=. --confcutdir=. \ + tests/test_session_memory_import_core.py ``` -Use the privacy-core command for privacy edits and the outbox-core command -for outbox edits. When changing the loader, source identity, or wiring around -either sibling, add the monolith identity regression: +Use the privacy-core command for privacy edits, the outbox-core command for +outbox edits, and the transcript-import command for transcript discovery or +selection edits. When changing the loader, source identity, or wiring around +any sibling, add the monolith identity regression: ```bash identity_pycache="$(mktemp -d "${TMPDIR:-/tmp}/aoa-session-memory-identity.XXXXXX")"