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/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")" 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..b077f0b0 --- /dev/null +++ b/scripts/aoa_session_memory_import.py @@ -0,0 +1,184 @@ +"""Small, read-only boundary for Codex transcript candidate selection. + +The session-memory producer owns enrichment and archive synchronization. This +module owns only the read-only part of historical import selection so +that collection policy can be tested without importing the full producer. +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 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 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, +) -> list[dict[str, Any]]: + """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 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() + 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(raw_path, source_stat, 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 + + +__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..2a04f812 --- /dev/null +++ b/tests/test_session_memory_import_core.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import importlib.util +import json +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 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" + 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, + transcript_probe=candidate_record, + transcript_size_prefilter_record=unexpected_size_prefilter, + 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, + transcript_size_prefilter_record=unexpected_size_prefilter, + ) + 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, + transcript_size_prefilter_record=unexpected_size_prefilter, + ) + assert second[0]["session_id"] == "bravo" + assert len(seen) == 2 + assert "bravo" in seen[1]