test_extract_session.py

import importlib.util
import json
import os
import subprocess
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location(
    "extract_session", ROOT / "scripts" / "extract_session.py"
)
extract_session = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
# Determinism pin (scoped + restored): DEFAULT_TIMEZONE is resolved at import, so
# pin the suite's default zone to +08:00 *for module exec only*, then restore the
# environment. This freezes the in-process defaults at +08:00 (deterministic date
# bucketing) WITHOUT leaking the override into subprocess.run children — so CLI/
# subprocess tests still exercise the real publish default ("local") and a local-
# default regression cannot hide. The publish default is also asserted
# directly via default_timezone({}).
_prev_extract_tz = os.environ.get("SESSION_EXTRACT_TZ")
os.environ["SESSION_EXTRACT_TZ"] = "+08:00"
SPEC.loader.exec_module(extract_session)
if _prev_extract_tz is None:
    os.environ.pop("SESSION_EXTRACT_TZ", None)
else:
    os.environ["SESSION_EXTRACT_TZ"] = _prev_extract_tz


def write_jsonl(path: Path, role: str, text: str, ts: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "type": role,
        "timestamp": ts,
        "message": {"role": role, "content": [{"type": "text", "text": text}]},
    }
    path.write_text(json.dumps(payload, ensure_ascii=False) + "\n", encoding="utf-8")


def append_jsonl(path: Path, payloads: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        "".join(json.dumps(payload, ensure_ascii=False) + "\n" for payload in payloads),
        encoding="utf-8",
    )


def test_process_all_groups_multiple_sessions_per_date(tmp_path):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"

    write_jsonl(
        sessions_dir / "first.jsonl",
        "user",
        "first session useful request",
        "2026-06-17T01:00:00Z",
    )
    write_jsonl(
        sessions_dir / "second.jsonl",
        "assistant",
        "second session useful answer that is long enough to avoid ack filtering",
        "2026-06-17T02:00:00Z",
    )

    extract_session.process_all(
        out_dir=out_dir,
        sessions_dir=sessions_dir,
        recursive=False,
    )

    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "**Sources:** 2 JSONL session file(s)" in output
    assert "Session `first.jsonl`" in output
    assert "Session `second.jsonl`" in output
    assert (out_dir / ".extract_session_state.json").exists()


def test_iter_session_files_recursive_is_explicit(tmp_path):
    sessions_dir = tmp_path / "sessions"
    write_jsonl(
        sessions_dir / "top.jsonl",
        "user",
        "top-level session",
        "2026-06-17T01:00:00Z",
    )
    write_jsonl(
        sessions_dir / "top" / "subagents" / "agent-a.jsonl",
        "assistant",
        "nested subagent session with enough text to survive filtering",
        "2026-06-17T02:00:00Z",
    )

    top_level = [p.relative_to(sessions_dir) for p in extract_session.iter_session_files(sessions_dir)]
    recursive = [
        p.relative_to(sessions_dir)
        for p in extract_session.iter_session_files(sessions_dir, recursive=True)
    ]

    assert top_level == [Path("top.jsonl")]
    assert Path("top.jsonl") in recursive
    assert Path("top/subagents/agent-a.jsonl") in recursive


def test_is_noise_does_not_drop_multiline_turn_with_ok_line():
    text = "Here is a useful note:\n\nok.\n\nContinue with the actual finding."
    assert not extract_session.is_noise(text)
    assert extract_session.is_noise("ok.")


def test_extract_text_blocks_string_strips_embedded_noise():
    raw = "Hello\n<system-reminder>INTERNAL</system-reminder>\nWorld"
    result = extract_session.extract_text_blocks(raw)
    assert "INTERNAL" not in result
    assert "Hello" in result
    assert "World" in result


def test_infer_date_default_bucket_uses_system_local_zone(tmp_path):
    # The default (no explicit tz) buckets by the system LOCAL zone — the publish
    # default — not a frozen owner offset. Pin TZ so this is deterministic.
    import time

    path = tmp_path / "session.jsonl"
    old_tz = os.environ.get("TZ")
    os.environ["TZ"] = "America/New_York"  # EST -05:00 in winter
    time.tzset()
    try:
        # 02:00Z Jan 2 is 21:00 EST Jan 1 → local default yields 2026-01-01.
        # The old +08:00 owner default would have yielded 2026-01-02.
        assert extract_session.infer_date([{"ts": "2026-01-02T02:00:00Z"}], path) == "2026-01-01"
    finally:
        if old_tz is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = old_tz
        time.tzset()


def test_default_timezone_defaults_to_local_not_owner_offset():
    # Publish contract: no owner-specific offset baked into the default.
    assert extract_session.default_timezone({}) == "local"
    # SESSION_EXTRACT_TZ still overrides.
    assert extract_session.default_timezone({"SESSION_EXTRACT_TZ": "UTC"}) == "UTC"
    assert extract_session.default_timezone({"SESSION_EXTRACT_TZ": "+05:30"}) == "+05:30"


def test_cli_default_timezone_is_local():
    # The published CLI default must be "local" regardless of the suite's in-process
    # +08:00 pin AND of any SESSION_EXTRACT_TZ the test runner/CI itself set. Strip
    # the override so this asserts the BINARY's default, not the environment
    # (a runner with the var preset must not flip this assertion).
    env = {k: v for k, v in os.environ.items() if k != "SESSION_EXTRACT_TZ"}
    result = subprocess.run(
        [sys.executable, str(ROOT / "scripts" / "extract_session.py"), "--help"],
        capture_output=True, text=True, env=env,
    )
    assert result.returncode == 0
    assert "default: local" in result.stdout
    assert "default: +08:00" not in result.stdout


def test_suite_tz_pin_is_restored_after_import():
    # The module-exec determinism pin must be restored, never left leaking in the
    # process env (and thus into subprocess children). Whatever
    # SESSION_EXTRACT_TZ was before the suite touched it is what it must be now.
    assert os.environ.get("SESSION_EXTRACT_TZ") == _prev_extract_tz


def test_default_sessions_dir_uses_env_override(monkeypatch, tmp_path):
    custom = tmp_path / "custom-sessions"
    monkeypatch.setenv("CLAUDE_SESSIONS_DIR", str(custom))
    assert extract_session.default_sessions_dir() == custom


def test_default_sessions_dir_supports_codex(monkeypatch, tmp_path):
    custom = tmp_path / "codex-sessions"
    monkeypatch.setenv("CODEX_SESSIONS_DIR", str(custom))
    assert extract_session.default_sessions_dir("codex") == custom


def test_parse_codex_jsonl_keeps_messages_and_drops_tools(tmp_path):
    # Uses real Codex rollout format: session_meta + response_item wrapper
    path = tmp_path / "rollout.jsonl"
    append_jsonl(
        path,
        [
            {"type": "session_meta", "timestamp": "2026-06-17T01:00:00Z", "payload": {}},
            {
                "type": "response_item",
                "timestamp": "2026-06-17T01:00:00Z",
                "payload": {
                    "type": "message",
                    "role": "user",
                    "content": [
                        {
                            "type": "input_text",
                            "text": "<environment_context>ignore</environment_context><user_instructions>ignore</user_instructions>",
                        }
                    ],
                },
            },
            {
                "type": "response_item",
                "timestamp": "2026-06-17T01:00:00Z",
                "payload": {
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": "mine this useful Codex request"}],
                },
            },
            {
                "type": "response_item",
                "payload": {"type": "function_call", "name": "shell", "arguments": "{}"},
            },
            {
                "type": "response_item",
                "timestamp": "2026-06-17T01:00:00Z",
                "payload": {
                    "type": "message",
                    "role": "assistant",
                    "content": [
                        {
                            "type": "output_text",
                            "text": "This is a useful Codex answer that is long enough to avoid ack filtering.",
                        }
                    ],
                },
            },
            {"type": "response_item", "payload": {"type": "reasoning", "summary": ["drop me"]}},
        ],
    )

    turns = extract_session.parse_jsonl(path, source="codex")

    assert [turn["role"] for turn in turns] == ["user", "assistant"]
    assert "mine this useful Codex request" in turns[0]["text"]
    assert "useful Codex answer" in turns[1]["text"]
    assert "environment_context" not in "\n".join(turn["text"] for turn in turns)
    assert "user_instructions" not in "\n".join(turn["text"] for turn in turns)
    assert all(turn["ts"] == "2026-06-17T01:00:00Z" for turn in turns)


def test_detect_source_for_codex_and_claude(tmp_path):
    codex = tmp_path / "codex.jsonl"
    append_jsonl(
        codex,
        [
            {"type": "session_meta", "timestamp": "2026-06-17T01:00:00Z", "payload": {}},
            {
                "type": "response_item",
                "timestamp": "2026-06-17T01:00:00Z",
                "payload": {
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": "hello"}],
                },
            },
        ],
    )
    claude = tmp_path / "claude.jsonl"
    write_jsonl(claude, "user", "hello from Claude", "2026-06-17T01:00:00Z")

    assert extract_session.detect_source(codex) == "codex"
    assert extract_session.detect_source(claude) == "claude"


def test_codex_batch_is_recursive_by_default(tmp_path):
    sessions_dir = tmp_path / "codex"
    out_dir = tmp_path / "out"
    append_jsonl(
        sessions_dir / "2026" / "06" / "17" / "rollout.jsonl",
        [
            {"type": "session_meta", "timestamp": "2026-06-17T01:00:00Z", "payload": {}},
            {
                "type": "response_item",
                "timestamp": "2026-06-17T01:00:00Z",
                "payload": {
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": "codex nested useful request"}],
                },
            },
        ],
    )

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="codex")

    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    state = json.loads((out_dir / ".extract_session_state.json").read_text(encoding="utf-8"))
    assert "codex nested useful request" in output
    assert state["source"] == "codex"
    assert state["recursive"] is True


def test_parse_timezone_supports_utc_and_offsets():
    turns = [{"ts": "2026-06-26T17:00:00Z"}]
    path = Path("unused.jsonl")
    assert extract_session.infer_date(turns, path, extract_session.parse_timezone("+08:00")) == "2026-06-27"
    assert extract_session.infer_date(turns, path, extract_session.parse_timezone("UTC")) == "2026-06-26"


def test_local_timezone_applies_per_timestamp_dst():
    """--timezone local must convert each timestamp with its own DST offset, not a
    single frozen current offset (historical offsets for local grouping)."""
    import os
    import time

    path = Path("unused.jsonl")
    old_tz = os.environ.get("TZ")
    os.environ["TZ"] = "America/New_York"
    time.tzset()
    try:
        tz = extract_session.parse_timezone("local")
        # 04:30Z in winter is EST (-05:00) → 23:30 the previous day
        winter = extract_session.infer_date([{"ts": "2026-01-01T04:30:00Z"}], path, tz)
        # 04:30Z in summer is EDT (-04:00) → 00:30 the same day
        summer = extract_session.infer_date([{"ts": "2026-07-01T04:30:00Z"}], path, tz)
        assert winter == "2025-12-31"
        assert summer == "2026-07-01"
    finally:
        if old_tz is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = old_tz
        time.tzset()


def test_process_all_second_run_skips_unchanged_sources(tmp_path, capsys):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "first.jsonl",
        "user",
        "first session useful request",
        "2026-06-17T01:00:00Z",
    )

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    first = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    capsys.readouterr()

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    second = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    err = capsys.readouterr().err

    assert first == second
    assert "[skip] 2026-06-17: unchanged" in err
    assert "nothing new or changed" in err


def test_process_all_rewrites_only_when_source_changes(tmp_path, capsys):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "first.jsonl"
    write_jsonl(path, "user", "first session useful request", "2026-06-17T01:00:00Z")

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    capsys.readouterr()
    write_jsonl(path, "user", "updated session useful request", "2026-06-17T01:00:00Z")

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    err = capsys.readouterr().err

    assert "updated session useful request" in output
    assert "[rewrite] 2026-06-17" in err


def test_process_all_period_filter(tmp_path):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "old.jsonl",
        "user",
        "old session useful request",
        "2026-06-16T01:00:00Z",
    )
    write_jsonl(
        sessions_dir / "new.jsonl",
        "user",
        "new session useful request",
        "2026-06-17T01:00:00Z",
    )

    extract_session.process_all(
        out_dir=out_dir,
        sessions_dir=sessions_dir,
        since="2026-06-17",
        until="2026-06-17",
    )

    assert not (out_dir / "2026-06-16.md").exists()
    assert (out_dir / "2026-06-17.md").exists()


def test_process_all_refuses_source_or_mode_mismatch_without_force(tmp_path):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "first.jsonl",
        "user",
        "first session useful request",
        "2026-06-17T01:00:00Z",
    )

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, recursive=False)

    try:
        extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, recursive=True)
    except ValueError as exc:
        assert "Use --force or a different --out" in str(exc)
    else:
        raise AssertionError("expected source/mode mismatch to fail")


def test_process_all_refuses_source_mismatch_without_force(tmp_path):
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "first.jsonl",
        "user",
        "first session useful request",
        "2026-06-17T01:00:00Z",
    )

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="claude")

    try:
        extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="codex")
    except ValueError as exc:
        assert "source='claude'" in str(exc)
    else:
        raise AssertionError("expected source mismatch to fail")


def test_cli_help_mentions_out_dir():
    result = subprocess.run(
        [sys.executable, str(ROOT / "scripts" / "extract_session.py"), "--help"],
        check=True,
        capture_output=True,
        text=True,
    )
    assert "--out-dir" in result.stdout
    assert "--source {claude,codex,openclaw,auto}" in result.stdout
    assert ".extract_session_state.json" in result.stdout


def test_parse_codex_jsonl_real_format(tmp_path):
    """parse_codex_jsonl must handle real Codex rollout format (response_item wrapper)."""
    path = tmp_path / "rollout.jsonl"
    append_jsonl(
        path,
        [
            {"type": "session_meta", "timestamp": "2026-06-26T00:27:36.194Z", "payload": {}},
            {
                "type": "response_item",
                "timestamp": "2026-06-26T00:27:36.213Z",
                "payload": {
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": "mine this useful Codex request"}],
                },
            },
            {
                "type": "response_item",
                "timestamp": "2026-06-26T00:27:37.000Z",
                "payload": {"type": "function_call", "name": "shell", "arguments": "{}"},
            },
            {
                "type": "response_item",
                "timestamp": "2026-06-26T00:27:38.000Z",
                "payload": {
                    "type": "message",
                    "role": "assistant",
                    "content": [
                        {
                            "type": "output_text",
                            "text": "This is a useful Codex answer that is long enough to avoid ack filtering.",
                        }
                    ],
                },
            },
        ],
    )

    turns = extract_session.parse_codex_jsonl(path)
    assert [t["role"] for t in turns] == ["user", "assistant"]
    assert "mine this useful Codex request" in turns[0]["text"]
    assert "useful Codex answer" in turns[1]["text"]


def test_batch_all_defaults_source_to_claude(tmp_path):
    """python3 extract_session.py --all must not fail when --source is omitted (defaults to claude)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "s.jsonl",
        "user",
        "useful session content for default source test",
        "2026-06-17T01:00:00Z",
    )
    # Invoke via CLI: default --source is "auto", main() should translate it to "claude" for --all
    result = subprocess.run(
        [
            sys.executable,
            str(ROOT / "scripts" / "extract_session.py"),
            "--all",
            "--timezone",
            "UTC",  # explicit: subprocess no longer inherits the suite's +08:00 pin
            "--sessions-dir",
            str(sessions_dir),
            "--out-dir",
            str(out_dir),
        ],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, f"Expected success, got stderr: {result.stderr}"
    assert (out_dir / "2026-06-17.md").exists()


def test_max_chars_change_triggers_rewrite(tmp_path):
    """Changing --max-chars must invalidate cached extracts."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "s.jsonl",
        "user",
        "useful session content for max chars test",
        "2026-06-17T01:00:00Z",
    )
    extract_session.process_all(max_chars=3000, out_dir=out_dir, sessions_dir=sessions_dir)
    try:
        extract_session.process_all(max_chars=100, out_dir=out_dir, sessions_dir=sessions_dir)
    except ValueError as exc:
        assert "Use --force or a different --out" in str(exc)
    else:
        raise AssertionError("expected max_chars mismatch to raise ValueError")


def test_deleted_source_file_removes_stale_extract(tmp_path):
    """Deleting a source JSONL must remove its content from the daily extract on the next run."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "s.jsonl"
    write_jsonl(path, "user", "useful session content for deletion test", "2026-06-17T01:00:00Z")

    # Run 1: normal extraction — creates the daily file and saves state
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert (out_dir / "2026-06-17.md").exists()

    # Delete the only source file
    path.unlink()

    # Run 2: normal re-run (no --force) — must detect deletion and remove the stale output
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-17.md").exists()


def test_session_moving_to_new_date_rebuilds_previous_date(tmp_path):
    """When a session's inferred date changes, the OLD date's extract must be
    rebuilt without it ('rebuild the previous date when a session moves')."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    b = sessions_dir / "b.jsonl"
    write_jsonl(a, "user", "alpha session text long enough to survive", "2026-06-17T01:00:00Z")
    write_jsonl(b, "user", "bravo session text long enough to survive", "2026-06-17T02:00:00Z")

    # Run 1: both land on the 17th
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    day17 = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "alpha session text" in day17 and "bravo session text" in day17

    # a.jsonl's timestamp moves to the 18th
    write_jsonl(a, "user", "alpha session text long enough to survive", "2026-06-18T01:00:00Z")

    # Run 2: the 18th gains alpha; the 17th must be rebuilt to drop alpha, keep bravo
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    day17 = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    day18 = (out_dir / "2026-06-18.md").read_text(encoding="utf-8")
    assert "alpha session text" in day18
    assert "bravo session text" in day17
    assert "alpha session text" not in day17  # stale copy must be gone


def test_changed_session_becoming_empty_removes_stale_extract(tmp_path):
    """A previously-indexed session that changes to only noise/tool content
    (zero useful turns) must have its stale extract removed
    (remove stale output when a changed session becomes empty)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "s.jsonl"
    write_jsonl(path, "user", "useful session content before going empty", "2026-06-17T01:00:00Z")

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert (out_dir / "2026-06-17.md").exists()

    # Rewrite the session to only a system-reminder (filtered to zero turns)
    append_jsonl(path, [{
        "type": "user",
        "timestamp": "2026-06-17T01:00:00Z",
        "message": {"role": "user", "content": [
            {"type": "text", "text": "<system-reminder>noise only</system-reminder>"}
        ]},
    }])

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-17.md").exists()


def test_changed_session_becoming_empty_is_not_reparsed_forever(tmp_path, capsys):
    """After a session becomes empty, its zero-turn result must be recorded in
    state so an unchanged re-run skips it instead of reparsing every time."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "s.jsonl"
    write_jsonl(path, "user", "useful session content before going empty", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)

    append_jsonl(path, [{
        "type": "user",
        "timestamp": "2026-06-17T01:00:00Z",
        "message": {"role": "user", "content": [
            {"type": "text", "text": "<system-reminder>noise only</system-reminder>"}
        ]},
    }])
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    capsys.readouterr()

    # Third run: file is unchanged since it went empty → must report nothing changed
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    err = capsys.readouterr().err
    assert "nothing new or changed" in err


def test_session_recovering_from_empty_is_re_extracted(tmp_path):
    """A file that went empty (recorded as a zero-turn marker) must be picked up
    again when it later gains real content."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "s.jsonl"
    write_jsonl(path, "user", "useful content before going empty", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)

    # Go empty (zero useful turns) → stale 17th removed, empty marker recorded
    append_jsonl(path, [{
        "type": "user",
        "timestamp": "2026-06-17T01:00:00Z",
        "message": {"role": "user", "content": [
            {"type": "text", "text": "<system-reminder>noise only</system-reminder>"}
        ]},
    }])
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-17.md").exists()

    # Recover with real content on a later date → must be extracted again
    write_jsonl(path, "user", "useful content after recovery", "2026-06-18T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    day18 = (out_dir / "2026-06-18.md").read_text(encoding="utf-8")
    assert "useful content after recovery" in day18


def test_session_moving_out_of_window_cleans_old_date(tmp_path):
    """A session moving from inside a --since/--until window to outside it must
    still rebuild/remove the in-window old date ('mark the old day
    dirty before skipping moved sessions')."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    write_jsonl(a, "user", "alpha text inside the window", "2026-06-17T01:00:00Z")

    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17", until="2026-06-17"
    )
    assert (out_dir / "2026-06-17.md").exists()

    # Move it to the 18th (outside the [17,17] window) and rerun with the same window
    write_jsonl(a, "user", "alpha text inside the window", "2026-06-18T01:00:00Z")
    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17", until="2026-06-17"
    )
    # The 17th had only this session, now moved out of window → its stale extract must go
    assert not (out_dir / "2026-06-17.md").exists()
    # And the 18th is outside the window → not produced
    assert not (out_dir / "2026-06-18.md").exists()


def test_out_of_window_deletion_is_cleaned_on_later_unfiltered_run(tmp_path):
    """Deleting a source whose date is outside the active --since window must not
    drop its state entry, so a later unfiltered run can still remove the stale
    extract ('preserve out-of-window deletions for later cleanup')."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    b = sessions_dir / "b.jsonl"
    write_jsonl(a, "user", "alpha on the sixteenth long enough", "2026-06-16T01:00:00Z")
    write_jsonl(b, "user", "bravo on the seventeenth long enough", "2026-06-17T01:00:00Z")

    # Run 1: unfiltered — both days created
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert (out_dir / "2026-06-16.md").exists()

    # Delete the 16th's source, then run with a window that EXCLUDES the 16th
    a.unlink()
    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17"
    )
    # 16th is out of window → not cleaned yet, but its state entry must be preserved
    assert (out_dir / "2026-06-16.md").exists()

    # Run 3: unfiltered again — the preserved entry lets the stale 16th be removed
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-16.md").exists()


def test_same_size_same_mtime_content_change_is_detected(tmp_path):
    """A content change with identical byte length AND identical mtime must still
    be detected (hash contents, don't trust mtime+size alone)."""
    import os
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    path = sessions_dir / "s.jsonl"
    # alpha / bravo are equal length, AAA / BBB are equal length → identical byte size
    write_jsonl(path, "user", "alpha content fixed length AAA", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    st = path.stat()

    write_jsonl(path, "user", "bravo content fixed length BBB", "2026-06-17T01:00:00Z")
    assert path.stat().st_size == st.st_size, "test setup: sizes must collide"
    os.utime(path, ns=(st.st_atime_ns, st.st_mtime_ns))  # restore mtime → full collision

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "bravo content" in output  # change detected despite identical mtime+size


def test_out_of_window_move_is_recorded_and_not_reparsed(tmp_path, monkeypatch):
    """A session that moves to a date outside the active window must be written
    back to state so repeated window-scoped runs skip it
    (no perpetual reparse for out-of-window moves)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    write_jsonl(a, "user", "alpha text inside the window", "2026-06-17T01:00:00Z")
    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17", until="2026-06-17"
    )
    # Move outside the window and rerun the same window (records the move)
    write_jsonl(a, "user", "alpha text inside the window", "2026-06-18T01:00:00Z")
    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17", until="2026-06-17"
    )

    # Third run, same window, file unchanged → it must NOT be reparsed
    calls = []
    real_parse = extract_session.parse_jsonl
    monkeypatch.setattr(
        extract_session, "parse_jsonl",
        lambda *a, **k: (calls.append(a[0]), real_parse(*a, **k))[1],
    )
    extract_session.process_all(
        out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17", until="2026-06-17"
    )
    assert calls == []  # settled out-of-window file is skipped, not reparsed


def test_out_of_window_move_between_out_of_window_dates_reconciles_later(tmp_path):
    """Moving a session between two dates that are BOTH outside the active window
    must not lose the old date's cleanup: a later unfiltered run removes the stale
    old extract and creates the new one (out-of-window move cleanup)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    write_jsonl(a, "user", "alpha on the sixteenth long enough", "2026-06-16T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)  # unfiltered
    assert (out_dir / "2026-06-16.md").exists()

    # Move to the 15th, then run a window that excludes BOTH the 15th and 16th
    write_jsonl(a, "user", "alpha on the fifteenth long enough", "2026-06-15T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-18")
    # The window run touched neither date; the 16th's stale extract still exists for now
    assert (out_dir / "2026-06-16.md").exists()

    # A later unfiltered run must reconcile both: drop the stale 16th, create the 15th
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-16.md").exists()
    assert (out_dir / "2026-06-15.md").exists()


def test_out_of_window_session_becoming_empty_reconciles_later(tmp_path):
    """A file outside the active window that changes to zero useful turns must not
    lose its old date's cleanup: a later unfiltered run removes the stale extract
    (out-of-window empty cleanup)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    write_jsonl(a, "user", "alpha on the sixteenth content", "2026-06-16T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)  # unfiltered
    assert (out_dir / "2026-06-16.md").exists()

    # Goes empty during a window that excludes the 16th
    append_jsonl(a, [{
        "type": "user",
        "timestamp": "2026-06-16T01:00:00Z",
        "message": {"role": "user", "content": [
            {"type": "text", "text": "<system-reminder>noise only</system-reminder>"}
        ]},
    }])
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-18")
    assert (out_dir / "2026-06-16.md").exists()  # not touched by the window run yet

    # Later unfiltered run reconciles: the now-empty source's stale extract is removed
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert not (out_dir / "2026-06-16.md").exists()


def test_out_of_window_content_edit_is_reconciled_later(tmp_path):
    """Editing a source whose date stays OUTSIDE the active window must still
    refresh that date's extract on a later unfiltered run, not leave stale text
    (mark out-of-window edits dirty)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    write_jsonl(a, "user", "original sixteenth content here", "2026-06-16T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)  # unfiltered
    assert "original sixteenth content" in (out_dir / "2026-06-16.md").read_text(encoding="utf-8")

    # Edit the 16th's content (same date), then run a window that EXCLUDES the 16th
    write_jsonl(a, "user", "REVISED sixteenth content present", "2026-06-16T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, since="2026-06-17")

    # A later unfiltered run must refresh 16.md with the revised content
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    text = (out_dir / "2026-06-16.md").read_text(encoding="utf-8")
    assert "REVISED sixteenth content" in text
    assert "original sixteenth content" not in text


def test_codex_agents_instructions_preamble_is_dropped(tmp_path):
    """The Codex-injected AGENTS `<INSTRUCTIONS>…</INSTRUCTIONS>` preamble (a
    role=user message) must be filtered out, not kept as Turn 1 in the extract
    (strip Codex injected AGENTS instructions)."""
    path = tmp_path / "rollout.jsonl"
    append_jsonl(path, [
        {"type": "session_meta", "timestamp": "2026-06-27T10:00:00Z", "payload": {}},
        {"type": "response_item", "timestamp": "2026-06-27T10:00:01Z", "payload": {
            "type": "message", "role": "user",
            "content": [{"type": "input_text",
                         "text": "# AGENTS.md instructions for /repo\n\n"
                                 "<INSTRUCTIONS>\n# OPERATIONS INDEX\nrepo rules here\n</INSTRUCTIONS>"}],
        }},
        {"type": "response_item", "timestamp": "2026-06-27T10:00:02Z", "payload": {
            "type": "message", "role": "user",
            "content": [{"type": "input_text",
                         "text": "real human question that is long enough to be kept here"}],
        }},
    ])
    turns = extract_session.parse_codex_jsonl(path)
    texts = [t["text"] for t in turns]
    assert not any("OPERATIONS INDEX" in x or "INSTRUCTIONS" in x for x in texts)
    assert any("real human question" in x for x in texts)


def test_codex_instructions_stripped_keeping_inline_prompt(tmp_path):
    """When the AGENTS preamble and the real request share one input_text, the
    instruction block is STRIPPED but the human prompt is preserved — not dropped
    as a whole turn (strip, don't drop the prompt)."""
    path = tmp_path / "rollout.jsonl"
    append_jsonl(path, [
        {"type": "session_meta", "timestamp": "2026-06-27T10:00:00Z", "payload": {}},
        {"type": "response_item", "timestamp": "2026-06-27T10:00:01Z", "payload": {
            "type": "message", "role": "user",
            "content": [{"type": "input_text",
                         "text": "# AGENTS.md instructions for /repo\n\n"
                                 "<INSTRUCTIONS>\nrepo ops rules here\n</INSTRUCTIONS>\n\n"
                                 "Now please refactor the parser for clarity"}],
        }},
    ])
    turns = extract_session.parse_codex_jsonl(path)
    texts = [t["text"] for t in turns]
    assert any("refactor the parser for clarity" in x for x in texts)
    assert not any("repo ops rules" in x or "INSTRUCTIONS" in x for x in texts)


def test_force_rebuild_removes_orphaned_extract_after_deletion(tmp_path):
    """--force after a source was deleted must still remove that source's stale
    extract, not orphan it (preserve deletion state during forced
    rebuilds)."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    a = sessions_dir / "a.jsonl"
    b = sessions_dir / "b.jsonl"
    write_jsonl(a, "user", "alpha on the sixteenth long enough", "2026-06-16T01:00:00Z")
    write_jsonl(b, "user", "bravo on the seventeenth long enough", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir)
    assert (out_dir / "2026-06-16.md").exists()

    a.unlink()
    # Forced rebuild must not leave the orphaned 16th behind
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, force=True)
    assert not (out_dir / "2026-06-16.md").exists()
    assert (out_dir / "2026-06-17.md").exists()


def test_cli_rejects_explicit_source_auto_in_batch(tmp_path):
    """Explicitly passing --source auto with --all must error, not silently run as
    claude (reject explicit batch --source auto)."""
    sessions_dir = tmp_path / "sessions"
    sessions_dir.mkdir()
    result = subprocess.run(
        [
            sys.executable, str(ROOT / "scripts" / "extract_session.py"),
            "--all", "--source", "auto",
            "--sessions-dir", str(sessions_dir),
            "--out-dir", str(tmp_path / "out"),
        ],
        capture_output=True, text=True,
    )
    assert result.returncode != 0
    assert "auto" in result.stderr.lower()


def test_cli_batch_without_source_defaults_to_claude(tmp_path):
    """Omitting --source in batch mode still defaults to claude (not rejected)."""
    sessions_dir = tmp_path / "sessions"
    write_jsonl(
        sessions_dir / "s.jsonl", "user", "useful default-source content", "2026-06-17T01:00:00Z"
    )
    result = subprocess.run(
        [
            sys.executable, str(ROOT / "scripts" / "extract_session.py"),
            "--all",
            "--timezone", "UTC",  # subprocess no longer inherits the suite's +08:00 pin
            "--sessions-dir", str(sessions_dir),
            "--out-dir", str(tmp_path / "out"),
        ],
        capture_output=True, text=True,
    )
    assert result.returncode == 0
    assert (tmp_path / "out" / "2026-06-17.md").exists()


def test_default_output_dir_and_files_are_private(tmp_path, monkeypatch):
    """The predictable default scratch dir (and its extracts) must be private:
    0700 dir, 0600 files — session logs can carry prompts/repo context/secrets."""
    import os
    import stat

    sessions_dir = tmp_path / "sessions"
    write_jsonl(
        sessions_dir / "s.jsonl", "user", "sensitive prompt content here", "2026-06-17T01:00:00Z"
    )
    fake_tmp = tmp_path / "tmproot"
    fake_tmp.mkdir()
    monkeypatch.setattr(extract_session.tempfile, "gettempdir", lambda: str(fake_tmp))
    monkeypatch.setattr(extract_session, "SCRATCH_DIR", fake_tmp / "session-extracts", raising=False)

    extract_session.process_all(out_dir=None, sessions_dir=sessions_dir)

    dirs = [p for p in fake_tmp.iterdir() if p.is_dir()]
    assert len(dirs) == 1, f"expected exactly one default scratch dir, got {dirs}"
    scratch = dirs[0]
    assert stat.S_IMODE(scratch.stat().st_mode) == 0o700, "default scratch dir must be 0700"

    md = scratch / "2026-06-17.md"
    assert md.exists()
    assert stat.S_IMODE(md.stat().st_mode) == 0o600, "extract .md must be 0600"
    state = scratch / ".extract_session_state.json"
    assert stat.S_IMODE(state.stat().st_mode) == 0o600, "state file must be 0600"


def test_default_dir_refuses_symlink(tmp_path, monkeypatch):
    """If the predictable default dir is pre-created as a symlink (tampering), the
    tool must refuse rather than follow it and write secrets to the target."""
    import os

    fake_tmp = tmp_path / "tmproot"
    fake_tmp.mkdir()
    monkeypatch.setattr(extract_session.tempfile, "gettempdir", lambda: str(fake_tmp))
    monkeypatch.setattr(extract_session, "SCRATCH_DIR", fake_tmp / "session-extracts", raising=False)

    target = tmp_path / "attacker-dir"
    target.mkdir()
    (fake_tmp / f"session-extracts-{os.getuid()}").symlink_to(target)

    sessions_dir = tmp_path / "sessions"
    write_jsonl(sessions_dir / "s.jsonl", "user", "secret content here", "2026-06-17T01:00:00Z")

    try:
        extract_session.process_all(out_dir=None, sessions_dir=sessions_dir)
    except ValueError as exc:
        assert "symlink" in str(exc).lower()
    else:
        raise AssertionError("expected refusal on a symlinked default scratch dir")


def test_cli_rejects_max_chars_below_two(tmp_path):
    """--max-chars 1 must be rejected: half = 1 // 2 = 0 makes truncate keep the
    full tail (text[-0:] == whole text), silently defeating the cap."""
    sessions_dir = tmp_path / "sessions"
    sessions_dir.mkdir()
    result = subprocess.run(
        [
            sys.executable, str(ROOT / "scripts" / "extract_session.py"),
            "--all", "--max-chars", "1",
            "--sessions-dir", str(sessions_dir), "--out-dir", str(tmp_path / "out"),
        ],
        capture_output=True, text=True,
    )
    assert result.returncode != 0
    assert "max-chars" in result.stderr.lower()


def test_process_all_rejects_max_chars_below_two(tmp_path):
    """The library entry point must also reject max_chars < 2 (privacy: a caller
    must not silently get the full tail when minimizing the extract)."""
    sessions_dir = tmp_path / "sessions"
    write_jsonl(sessions_dir / "s.jsonl", "user", "content here long enough", "2026-06-17T01:00:00Z")
    try:
        extract_session.process_all(max_chars=1, out_dir=tmp_path / "out", sessions_dir=sessions_dir)
    except ValueError as exc:
        assert "max_chars" in str(exc).lower() or "max-chars" in str(exc).lower()
    else:
        raise AssertionError("expected process_all to reject max_chars < 2")


def test_local_timezone_state_is_keyed_by_resolved_zone(tmp_path):
    """A state dir created with --timezone local under one host zone must not be
    silently reused under a different host zone — the cached dates would be stale
    (key local state by resolved zone)."""
    import os
    import time

    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(sessions_dir / "s.jsonl", "user", "content here long enough", "2026-06-17T01:00:00Z")

    old_tz = os.environ.get("TZ")
    os.environ["TZ"] = "America/New_York"
    time.tzset()
    try:
        extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, timezone_value="local")
        os.environ["TZ"] = "Asia/Tokyo"
        time.tzset()
        try:
            extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, timezone_value="local")
        except ValueError as exc:
            assert "timezone" in str(exc).lower() or "force" in str(exc).lower()
        else:
            raise AssertionError("reusing a local-tz state dir under a different zone must be refused")
    finally:
        if old_tz is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = old_tz
        time.tzset()


def test_truncate_rejects_max_chars_below_two():
    """truncate() itself must reject max_chars < 2 (half = 0 → text[-0:] leaks the
    whole tail), so a direct caller can't bypass the guard."""
    try:
        extract_session.truncate("abcdef", 1)
    except ValueError:
        pass
    else:
        raise AssertionError("truncate must reject max_chars < 2")


def test_parse_ymd_normalizes_zero_padding():
    """parse_ymd must return a zero-padded canonical date so lexicographic range
    comparison against inferred dates is correct."""
    assert extract_session.parse_ymd("2026-6-7", "--since") == "2026-06-07"
    assert extract_session.parse_ymd("2026-06-07", "--since") == "2026-06-07"
    assert extract_session.parse_ymd(None, "--since") is None


def test_non_zero_padded_since_does_not_exclude_matching_dates(tmp_path):
    """--since 2026-6-7 must NOT silently exclude 2026-06-17 (raw unpadded string
    sorts after the padded inferred date)'s exact example."""
    sessions_dir = tmp_path / "sessions"
    out_dir = tmp_path / "out"
    write_jsonl(
        sessions_dir / "a.jsonl", "user", "seventeenth content long enough here", "2026-06-17T01:00:00Z"
    )
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, since="2026-6-7")
    assert (out_dir / "2026-06-17.md").exists()


# --- Codex --project filter + coverage --list (pre-publish hardening) ---

def _codex_meta(payload: dict) -> dict:
    return {"type": "session_meta", "timestamp": "2026-06-17T01:00:00Z", "payload": payload}


def _codex_msg(role: str, text: str, ts: str = "2026-06-17T01:00:00Z") -> dict:
    tt = "input_text" if role == "user" else "output_text"
    return {
        "type": "response_item",
        "timestamp": ts,
        "payload": {"type": "message", "role": role, "content": [{"type": tt, "text": text}]},
    }


def test_session_project_codex_new_format_from_git_url(tmp_path):
    p = tmp_path / "s.jsonl"
    append_jsonl(p, [
        _codex_meta({"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:owner/astro.git"}}),
        _codex_msg("user", "hi there long enough to keep"),
    ])
    assert extract_session.session_project(p, source="codex") == "astro"


def test_session_project_codex_new_format_falls_back_to_cwd(tmp_path):
    p = tmp_path / "s.jsonl"
    append_jsonl(p, [
        _codex_meta({"cwd": "/home/u/air"}),  # no git
        _codex_msg("user", "hi there long enough to keep"),
    ])
    assert extract_session.session_project(p, source="codex") == "air"


def test_session_project_codex_legacy_header_git_url(tmp_path):
    p = tmp_path / "s.jsonl"
    p.write_text(
        json.dumps({"id": "x", "timestamp": "2026-06-17T01:00:00Z", "instructions": "AGENTS",
                    "git": {"repository_url": "git@github.com:owner/astro.git"}}) + "\n"
        + json.dumps({"type": "message", "role": "user",
                      "content": [{"type": "text", "text": "legacy hello long enough"}]}) + "\n",
        encoding="utf-8",
    )
    assert extract_session.session_project(p, source="codex") == "astro"


def test_session_project_none_when_unknown(tmp_path):
    p = tmp_path / "s.jsonl"
    append_jsonl(p, [_codex_meta({}), _codex_msg("user", "no project info here but long enough")])
    assert extract_session.session_project(p, source="codex") is None


def test_process_all_project_filter_excludes_other_repo(tmp_path):
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    append_jsonl(sessions / "astro.jsonl", [
        _codex_meta({"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:owner/astro.git"}}),
        _codex_msg("user", "astro project content long enough to keep"),
    ])
    append_jsonl(sessions / "air.jsonl", [
        _codex_meta({"cwd": "/home/u/air", "git": {"repository_url": "git@github.com:owner/air.git"}}),
        _codex_msg("user", "air game content long enough to keep"),
    ])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex", project="astro")
    output = (out / "2026-06-17.md").read_text(encoding="utf-8")
    assert "astro project content" in output
    assert "air game content" not in output


def test_list_sessions_reports_project_and_turns(tmp_path, capsys):
    sessions = tmp_path / "codex"
    append_jsonl(sessions / "astro.jsonl", [
        _codex_meta({"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:owner/astro.git"}}),
        _codex_msg("user", "astro content long enough to keep here"),
    ])
    append_jsonl(sessions / "air.jsonl", [
        _codex_meta({"cwd": "/home/u/air"}),
        _codex_msg("user", "air content long enough to keep here"),
    ])
    extract_session.list_sessions(sessions_dir=sessions, source="codex")
    out = capsys.readouterr().out
    assert "astro" in out and "air" in out
    assert "2026-06-17" in out


def test_list_sessions_respects_project_filter(tmp_path, capsys):
    sessions = tmp_path / "codex"
    append_jsonl(sessions / "astro.jsonl", [
        _codex_meta({"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:owner/astro.git"}}),
        _codex_msg("user", "astro content long enough to keep here"),
    ])
    append_jsonl(sessions / "air.jsonl", [
        _codex_meta({"cwd": "/home/u/air"}),
        _codex_msg("user", "air content long enough to keep here"),
    ])
    extract_session.list_sessions(sessions_dir=sessions, source="codex", project="astro")
    out = capsys.readouterr().out
    assert "astro" in out
    assert "air.jsonl" not in out


def test_list_extracted_status_uses_per_session_state(tmp_path, capsys):
    # A new session on an already-extracted DATE must read extracted=no — date-file
    # existence is too coarse; coverage must be per-session (state signature).
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    append_jsonl(sessions / "astro.jsonl", [
        _codex_meta(ag), _codex_msg("user", "first astro session long enough to keep"),
    ])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex")
    # add a NEW session on the SAME date, not yet extracted
    append_jsonl(sessions / "astro2.jsonl", [
        _codex_meta(ag), _codex_msg("user", "second astro session long enough", "2026-06-17T05:00:00Z"),
    ])
    extract_session.list_sessions(sessions_dir=sessions, source="codex", out_dir=out)
    lines = capsys.readouterr().out.splitlines()
    row_old = next(l for l in lines if l.endswith("astro.jsonl"))
    row_new = next(l for l in lines if l.endswith("astro2.jsonl"))
    assert "yes" in row_old          # the extracted session
    assert "no" in row_new           # new, not yet in the extract (despite the date file existing)


def test_process_all_rejects_project_filter_change_on_shared_outdir(tmp_path):
    # The corruption guard: reusing one out-dir across different --project scopes (incl.
    # unfiltered -> astro) must be refused, else other-project extracts get rewritten.
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    air = {"cwd": "/home/u/air", "git": {"repository_url": "git@github.com:o/air.git"}}
    append_jsonl(sessions / "astro.jsonl", [_codex_meta(ag), _codex_msg("user", "astro content long enough")])
    append_jsonl(sessions / "air.jsonl", [_codex_meta(air), _codex_msg("user", "air content long enough")])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex")  # unfiltered
    try:
        extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex", project="astro")
    except ValueError as exc:
        assert "project=" in str(exc)
    else:
        raise AssertionError("expected project-filter change on shared out-dir to fail")


def test_list_extracted_status_excludes_out_of_window_session(tmp_path, capsys):
    # The state records a signature for OUT-OF-WINDOW sessions too (deferred via
    # pending_dirty) without writing their .md — those must read extracted=no.
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    append_jsonl(sessions / "d1.jsonl", [
        _codex_meta(ag), _codex_msg("user", "day one content long enough", "2026-06-17T01:00:00Z")])
    append_jsonl(sessions / "d2.jsonl", [
        _codex_meta(ag), _codex_msg("user", "day two content long enough", "2026-06-18T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex", until="2026-06-17")
    extract_session.list_sessions(sessions_dir=sessions, source="codex", out_dir=out)
    lines = capsys.readouterr().out.splitlines()
    assert "yes" in next(l for l in lines if l.endswith("d1.jsonl"))   # in window, .md written
    assert "no" in next(l for l in lines if l.endswith("d2.jsonl"))    # out of window, no .md


def test_process_all_rejects_legacy_unscoped_state_under_project_filter(tmp_path):
    # A state written before --project existed has files but no `project` field. Applying
    # --project to that out-dir would silently treat other-project entries as deleted —
    # so it must be refused (legacy non-empty state = old unfiltered scope, not a wildcard).
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    air = {"cwd": "/home/u/air", "git": {"repository_url": "git@github.com:o/air.git"}}
    append_jsonl(sessions / "astro.jsonl", [_codex_meta(ag), _codex_msg("user", "astro content long enough")])
    append_jsonl(sessions / "air.jsonl", [_codex_meta(air), _codex_msg("user", "air content long enough")])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex")
    sp = out / ".extract_session_state.json"
    st = json.loads(sp.read_text(encoding="utf-8"))
    del st["project"]                       # simulate a pre-feature (legacy) state
    sp.write_text(json.dumps(st), encoding="utf-8")
    try:
        extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex", project="astro")
    except ValueError as exc:
        assert "project=" in str(exc)
    else:
        raise AssertionError("expected legacy unscoped state + --project to be refused")


def test_list_resolves_default_out_dir_when_none(tmp_path, capsys, monkeypatch):
    # `--all --list` with no --out-dir must report coverage of the DEFAULT scratch dir
    # (where `--all` without --out-dir extracts), not blanket EXTRACTED=no.
    sessions = tmp_path / "codex"
    defdir = tmp_path / "default-scratch"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    append_jsonl(sessions / "astro.jsonl", [_codex_meta(ag), _codex_msg("user", "astro content long enough")])
    extract_session.process_all(out_dir=defdir, sessions_dir=sessions, source="codex")
    monkeypatch.setattr(extract_session, "default_scratch_dir", lambda: defdir)
    extract_session.list_sessions(sessions_dir=sessions, source="codex", out_dir=None)
    lines = capsys.readouterr().out.splitlines()
    assert "yes" in next(l for l in lines if l.endswith("astro.jsonl"))


def test_list_marks_coverage_unknown_on_scope_mismatch(tmp_path, capsys):
    # --list against an out-dir whose state was produced with DIFFERENT options (e.g.
    # max_chars) must not claim EXTRACTED=yes — process_all would reject that state.
    sessions = tmp_path / "codex"
    out = tmp_path / "out"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    append_jsonl(sessions / "astro.jsonl", [_codex_meta(ag), _codex_msg("user", "astro content long enough to keep")])
    extract_session.process_all(out_dir=out, sessions_dir=sessions, source="codex", max_chars=10)
    extract_session.list_sessions(sessions_dir=sessions, source="codex", out_dir=out, max_chars=3000)
    row = next(l for l in capsys.readouterr().out.splitlines() if l.endswith("astro.jsonl"))
    assert "yes" not in row
    assert "?" in row


def test_list_rejects_inverted_date_range(tmp_path):
    # --list must reject --since > --until like process_all does; otherwise a typo yields
    # an empty manifest and users skip days (the exact failure --list is meant to prevent).
    sessions = tmp_path / "codex"
    ag = {"cwd": "/home/u/astro", "git": {"repository_url": "git@github.com:o/astro.git"}}
    append_jsonl(sessions / "astro.jsonl", [_codex_meta(ag), _codex_msg("user", "astro content long enough")])
    try:
        extract_session.list_sessions(sessions_dir=sessions, source="codex",
                                      since="2026-06-20", until="2026-06-10")
    except ValueError as exc:
        assert "must be <=" in str(exc)
    else:
        raise AssertionError("expected inverted --list date range to be rejected")


# --- OpenClaw source adapter ---


def _openclaw_session(cwd: str = "/home/u/clawria-workspace", ts: str = "2026-06-17T01:00:00Z") -> dict:
    return {"type": "session", "version": 1, "id": "sess-1", "timestamp": ts, "cwd": cwd}


def _openclaw_msg(role: str, text: str, ts: str = "2026-06-17T01:00:00Z", extra_blocks=None) -> dict:
    content = list(extra_blocks or []) + [{"type": "text", "text": text}]
    return {"type": "message", "timestamp": ts, "message": {"role": role, "content": content}}


def test_default_sessions_dir_supports_openclaw(monkeypatch, tmp_path):
    custom = tmp_path / "openclaw-sessions"
    monkeypatch.setenv("OPENCLAW_SESSIONS_DIR", str(custom))
    assert extract_session.default_sessions_dir("openclaw") == custom


def test_default_sessions_dir_openclaw_default(monkeypatch):
    monkeypatch.delenv("OPENCLAW_SESSIONS_DIR", raising=False)
    monkeypatch.delenv("SESSION_EXTRACT_DIR", raising=False)
    expected = Path.home() / ".openclaw/agents/main/sessions"
    assert extract_session.default_sessions_dir("openclaw") == expected


def test_detect_source_for_openclaw_with_header(tmp_path):
    p = tmp_path / "oc.jsonl"
    append_jsonl(p, [_openclaw_session(), _openclaw_msg("user", "hello from OpenClaw long enough to keep")])
    assert extract_session.detect_source(p) == "openclaw"


def test_detect_source_for_openclaw_headerless(tmp_path):
    # a resumed/partial file with no session header still detects via nested message.role
    p = tmp_path / "oc.jsonl"
    append_jsonl(p, [_openclaw_msg("assistant", "answer text long enough to avoid ack filtering")])
    assert extract_session.detect_source(p) == "openclaw"


def test_parse_openclaw_jsonl_keeps_text_drops_thinking_and_tools(tmp_path):
    path = tmp_path / "oc.jsonl"
    append_jsonl(path, [
        _openclaw_session(),
        _openclaw_msg("user", "mine this useful OpenClaw request"),
        _openclaw_msg(
            "assistant",
            "This is a useful OpenClaw answer long enough to avoid ack filtering.",
            extra_blocks=[
                {"type": "thinking", "text": "internal reasoning DROP ME"},
                {"type": "toolCall", "name": "bashtool", "arguments": {"command": "ls"}},
            ],
        ),
    ])

    turns = extract_session.parse_jsonl(path, source="openclaw")

    assert [t["role"] for t in turns] == ["user", "assistant"]
    assert "mine this useful OpenClaw request" in turns[0]["text"]
    assert "useful OpenClaw answer" in turns[1]["text"]
    joined = "\n".join(t["text"] for t in turns)
    assert "DROP ME" not in joined
    assert "bashtool" not in joined
    assert all(t["ts"] == "2026-06-17T01:00:00Z" for t in turns)


def test_parse_openclaw_user_string_content(tmp_path):
    # OpenClaw cron/system prompts arrive as a plain string, not a list of blocks.
    path = tmp_path / "oc.jsonl"
    append_jsonl(path, [
        _openclaw_session(),
        {"type": "message", "timestamp": "2026-06-17T01:00:00Z",
         "message": {"role": "user", "content": "plain string prompt long enough to keep"}},
    ])
    turns = extract_session.parse_jsonl(path, source="openclaw")
    assert len(turns) == 1
    assert "plain string prompt" in turns[0]["text"]


def test_parse_openclaw_drops_tool_only_turn(tmp_path):
    path = tmp_path / "oc.jsonl"
    append_jsonl(path, [
        _openclaw_session(),
        {"type": "message", "timestamp": "2026-06-17T01:00:00Z",
         "message": {"role": "assistant", "content": [
             {"type": "toolCall", "name": "bashtool", "arguments": {}},
             {"type": "thinking", "text": "no user-facing text"},
         ]}},
    ])
    assert extract_session.parse_jsonl(path, source="openclaw") == []


def test_openclaw_batch_top_level_not_recursive(tmp_path):
    sessions_dir = tmp_path / "oc"
    out_dir = tmp_path / "out"
    append_jsonl(sessions_dir / "main.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "openclaw top-level useful request")])
    append_jsonl(sessions_dir / "nested" / "deep.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "nested should be ignored by default")])

    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="openclaw")

    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "Session `main.jsonl`" in output
    assert "nested should be ignored" not in output


def test_session_project_openclaw_from_cwd(tmp_path):
    p = tmp_path / "oc.jsonl"
    append_jsonl(p, [_openclaw_session(cwd="/home/u/clawria-workspace"),
                     _openclaw_msg("user", "hi there long enough to keep")])
    assert extract_session.session_project(p, source="openclaw") == "clawria-workspace"


def test_openclaw_iter_excludes_checkpoint_files(tmp_path):
    # OpenClaw writes branch/restore SNAPSHOTS as `<id>.checkpoint.<uuid>.jsonl` — derived
    # artifacts that duplicate the base session and carry an internal compaction forest.
    # iter_session_files must skip them for the openclaw source; the live `<id>.jsonl` is the
    # active linear history. (Real-data note: forks appear ONLY in checkpoint files.)
    sessions_dir = tmp_path / "oc"
    append_jsonl(sessions_dir / "13dd417a.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "live base session useful request")])
    append_jsonl(sessions_dir / "13dd417a.checkpoint.51cf22ed.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "checkpoint snapshot must be excluded")])

    names = [p.name for p in extract_session.iter_session_files(sessions_dir, source="openclaw")]
    assert names == ["13dd417a.jsonl"]

    # end-to-end: the batch extract never includes the checkpoint snapshot
    out_dir = tmp_path / "out"
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="openclaw")
    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "live base session" in output
    assert "checkpoint snapshot" not in output


def test_openclaw_iter_excludes_trajectory_sidecars(tmp_path):
    # OpenClaw run/trajectory artifacts `<id>.trajectory.jsonl` sit beside the live `<id>.jsonl`
    # and would otherwise be scanned (hashed/parsed as bogus sessions, or — if message-shaped —
    # pollute/duplicate the extract). iter_session_files must skip them for the openclaw source.
    sessions_dir = tmp_path / "oc"
    append_jsonl(sessions_dir / "13dd417a.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "live base session useful request")])
    append_jsonl(sessions_dir / "13dd417a.trajectory.jsonl", [
        _openclaw_session(), _openclaw_msg("user", "trajectory sidecar must be excluded")])

    names = [p.name for p in extract_session.iter_session_files(sessions_dir, source="openclaw")]
    assert names == ["13dd417a.jsonl"]

    out_dir = tmp_path / "out"
    extract_session.process_all(out_dir=out_dir, sessions_dir=sessions_dir, source="openclaw")
    output = (out_dir / "2026-06-17.md").read_text(encoding="utf-8")
    assert "live base session" in output
    assert "trajectory sidecar" not in output


def test_list_sessions_openclaw_manifest(tmp_path, capsys):
    sessions_dir = tmp_path / "oc"
    append_jsonl(sessions_dir / "s1.jsonl", [
        _openclaw_session(cwd="/home/u/clawria-workspace"),
        _openclaw_msg("user", "openclaw list manifest useful request")])
    extract_session.list_sessions(sessions_dir=sessions_dir, source="openclaw")
    out = capsys.readouterr().out
    assert "2026-06-17" in out
    assert "clawria-workspace" in out
    assert "s1.jsonl" in out


def test_parse_tolerates_invalid_utf8_bytes(tmp_path):
    # A non-UTF-8 / corrupt byte in one line must not abort the whole parse — errors="replace"
    # keeps the read going so the valid turns still extract.
    p = tmp_path / "s.jsonl"
    good = json.dumps({"type": "user", "message": {"role": "user",
                       "content": [{"type": "text", "text": "valid useful request here"}]}})
    p.write_bytes(good.encode("utf-8") + b"\n" + b"\xff\xfe not valid utf8 garbage line\n")
    turns = extract_session.parse_jsonl(p, source="claude")   # must not raise UnicodeDecodeError
    assert any("valid useful request" in t["text"] for t in turns)


def test_cli_version_flag(tmp_path):
    result = subprocess.run(
        [sys.executable, str(ROOT / "scripts" / "extract_session.py"), "--version"],
        capture_output=True, text=True,
    )
    assert result.returncode == 0
    assert extract_session.__version__ in result.stdout


# --- OpenClaw checkpoint deep-history recovery (--checkpoints, default off) ---


def _oc_checkpoint(path, turns, cwd="/home/u/clawria-workspace"):
    """turns: list of (role, text, ts) — write an <id>.checkpoint.<uuid>.jsonl snapshot."""
    payloads = [_openclaw_session(cwd=cwd)]
    for role, text, ts in turns:
        payloads.append(_openclaw_msg(role, text, ts))
    append_jsonl(path, payloads)


def test_openclaw_checkpoints_ignored_by_default(tmp_path):
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl",
                 [_openclaw_session(), _openclaw_msg("user", "base recent turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl",
                   [("user", "DEEP history turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw")  # no flag
    assert (out / "2026-06-17.md").exists()
    assert not (out / "2026-06-15.md").exists()   # checkpoint deep history NOT recovered by default


def test_openclaw_checkpoints_recover_deep_history(tmp_path):
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl",
                 [_openclaw_session(), _openclaw_msg("user", "base recent turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [
        ("user", "DEEP early turn long enough to keep", "2026-06-15T01:00:00Z"),
        ("assistant", "deep assistant reply long enough to avoid ack filtering", "2026-06-15T01:01:00Z"),
    ])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert "DEEP early turn" in (out / "2026-06-15.md").read_text(encoding="utf-8")
    assert "base recent turn" in (out / "2026-06-17.md").read_text(encoding="utf-8")


def test_openclaw_checkpoints_bucket_by_own_date(tmp_path):
    # turns spanning two days must land in two day files (NOT all under the first date)
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl",
                 [_openclaw_session(), _openclaw_msg("user", "base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [
        ("user", "turn on the fifteenth long enough to keep", "2026-06-15T01:00:00Z"),
        ("user", "turn on the sixteenth long enough to keep", "2026-06-16T01:00:00Z"),
    ])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert "fifteenth" in (out / "2026-06-15.md").read_text(encoding="utf-8")
    assert "sixteenth" in (out / "2026-06-16.md").read_text(encoding="utf-8")
    assert "sixteenth" not in (out / "2026-06-15.md").read_text(encoding="utf-8")


def test_openclaw_checkpoints_dedup_no_double_count(tmp_path):
    # a turn present in BOTH base and checkpoint, and duplicated WITHIN the checkpoint, appears once
    sd = tmp_path / "oc"; out = tmp_path / "out"
    shared = ("user", "shared turn present in base and checkpoint long enough", "2026-06-17T01:00:00Z")
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(), _openclaw_msg(*shared)])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [shared, shared,  # intra-checkpoint dup
                   ("user", "deep unique turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    day17 = (out / "2026-06-17.md").read_text(encoding="utf-8")
    assert day17.count("shared turn present in base and checkpoint") == 1   # not doubled/tripled
    assert "deep unique turn" in (out / "2026-06-15.md").read_text(encoding="utf-8")


def test_openclaw_checkpoints_union_all_dedup(tmp_path):
    # Union ALL checkpoints (newest->oldest), dedup by signature — NOT latest-only.
    # A size-capped 'latest' snapshot could have trimmed the oldest turns; the older
    # checkpoint's unique tail must still be recovered, and shared turns appear once.
    import os
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(), _openclaw_msg("user", "base turn long enough to keep", "2026-06-17T01:00:00Z")])
    shared = ("user", "shared deep turn present in both checkpoints long enough", "2026-06-15T01:00:00Z")
    older = sd / "sess.checkpoint.old.jsonl"; newer = sd / "sess.checkpoint.new.jsonl"
    _oc_checkpoint(older, [("user", "OLDEST unique turn the latest may have trimmed long", "2026-06-13T01:00:00Z"), shared])
    _oc_checkpoint(newer, [shared, ("user", "NEWER deep turn long enough to keep", "2026-06-16T01:00:00Z")])
    os.utime(older, (1000, 1000)); os.utime(newer, (2000, 2000))
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    # the oldest unique turn (which a size-capped latest could have dropped) is still recovered:
    assert "OLDEST unique turn" in (out / "2026-06-13.md").read_text(encoding="utf-8")
    assert "NEWER deep turn" in (out / "2026-06-16.md").read_text(encoding="utf-8")
    # shared turn deduped to exactly one occurrence across the whole union:
    assert (out / "2026-06-15.md").read_text(encoding="utf-8").count("shared deep turn present in both checkpoints") == 1


def test_openclaw_checkpoints_signature_uses_raw_not_truncated(tmp_path):
    # Two DISTINCT turns sharing head+tail but differing only in the (truncated-away) middle
    # must NOT merge: the dedup signature is computed on RAW text, before max_chars truncation.
    sd = tmp_path / "oc"; out = tmp_path / "out"
    head = "HEAD common prefix long enough to matter "; tail = " TAIL common suffix"
    base_text = head + ("A" * 400) + tail
    ckpt_text = head + ("B" * 400) + tail   # same head+tail, different middle
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(), _openclaw_msg("user", base_text, "2026-06-15T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [("user", ckpt_text, "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True, max_chars=60)
    day = (out / "2026-06-15.md").read_text(encoding="utf-8")
    # both distinct turns survive (raw signatures differ even though truncated renderings match):
    assert day.count("HEAD common prefix") == 2


def test_openclaw_checkpoints_no_inflation_when_pure_duplicate(tmp_path):
    # a checkpoint whose only turn already exists in the base contributes nothing new
    sd = tmp_path / "oc"; out = tmp_path / "out"
    only = ("user", "the only turn there is long enough to keep", "2026-06-17T01:00:00Z")
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(), _openclaw_msg(*only)])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [only])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert (out / "2026-06-17.md").read_text(encoding="utf-8").count("the only turn there is") == 1
    assert list(out.glob("*.md")) == [out / "2026-06-17.md"]   # no spurious extra day files


def test_openclaw_checkpoints_still_skip_trajectory(tmp_path):
    # even with recovery on, .trajectory.jsonl is never read
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(), _openclaw_msg("user", "base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.trajectory.jsonl", [("user", "trajectory turn must be ignored long enough", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert not (out / "2026-06-15.md").exists()


def test_openclaw_checkpoints_respect_project_filter(tmp_path):
    # --project must NOT leak another project's checkpoint turns into the recovery.
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "a.jsonl", [_openclaw_session(cwd="/home/u/astro"),
                 _openclaw_msg("user", "astro base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "a.checkpoint.x.jsonl",
                   [("user", "astro deep turn long enough to keep", "2026-06-15T01:00:00Z")], cwd="/home/u/astro")
    append_jsonl(sd / "b.jsonl", [_openclaw_session(cwd="/home/u/other"),
                 _openclaw_msg("user", "other base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "b.checkpoint.y.jsonl",
                   [("user", "OTHER deep turn must be excluded long enough", "2026-06-15T01:00:00Z")], cwd="/home/u/other")
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw",
                                recover_checkpoints=True, project="astro")
    day15 = (out / "2026-06-15.md").read_text(encoding="utf-8")
    assert "astro deep turn" in day15
    assert "OTHER deep turn" not in day15


def test_openclaw_checkpoints_incremental_rerun_keeps_base_and_recovers(tmp_path):
    # extract normally first, then re-run with --checkpoints on the SAME out-dir: the base day
    # must keep its content (the placeholder re-parse) AND gain the recovered deep turns.
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(),
                 _openclaw_msg("user", "base turn that must survive the rerun long enough", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl", [
        ("user", "same-day deep turn long enough to keep", "2026-06-17T02:00:00Z"),
        ("user", "earlier-day deep turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw")  # run 1: normal
    assert "base turn that must survive" in (out / "2026-06-17.md").read_text(encoding="utf-8")
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)  # run 2
    day17 = (out / "2026-06-17.md").read_text(encoding="utf-8")
    assert "base turn that must survive" in day17        # base content not lost on the rewrite
    assert "same-day deep turn" in day17                  # recovered turn merged into the base day
    assert "earlier-day deep turn" in (out / "2026-06-15.md").read_text(encoding="utf-8")


def test_openclaw_checkpoints_warns_for_non_openclaw_source(tmp_path, capsys):
    sd = tmp_path / "cl"; out = tmp_path / "out"
    write_jsonl(sd / "s.jsonl", "user", "claude session turn long enough to keep", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="claude", recover_checkpoints=True)
    assert "--checkpoints is openclaw-only" in capsys.readouterr().err


def test_openclaw_checkpoints_remove_stale_recovered_day(tmp_path):
    # E: a checkpoint-only recovered day whose checkpoint later disappears
    # must have its stale .md removed on the next --checkpoints run — no live session reconciles it
    # (key=None left no state entry / pending_dirty date). This is the REMOVAL direction, not augment.
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(),
                 _openclaw_msg("user", "base recent turn long enough to keep", "2026-06-17T01:00:00Z")])
    ck = sd / "sess.checkpoint.aaaa.jsonl"
    _oc_checkpoint(ck, [("user", "deep checkpoint-only day turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert (out / "2026-06-15.md").exists()        # recovered checkpoint-only day written
    ck.unlink()                                     # the checkpoint disappears
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert not (out / "2026-06-15.md").exists()     # stale recovered-only day removed
    assert (out / "2026-06-17.md").exists()         # the live-session day is untouched


def test_openclaw_checkpoints_augment_shrink_rewrites_live_day(tmp_path):
    # Complement to E: a day with a LIVE session that ALSO got recovered
    # turns; the checkpoint later disappears. The live session is unchanged (would [skip]), so a
    # re-run must FORCE a rewrite to drop the now-stale recovered section — without deleting the day.
    sd = tmp_path / "oc"; out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [_openclaw_session(),
                 _openclaw_msg("user", "live base turn stays long enough to keep", "2026-06-17T01:00:00Z")])
    ck = sd / "sess.checkpoint.aaaa.jsonl"
    _oc_checkpoint(ck, [("user", "recovered augment turn for the same day long enough", "2026-06-17T02:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    day = (out / "2026-06-17.md").read_text(encoding="utf-8")
    assert "live base turn stays" in day and "recovered augment turn" in day   # augmented
    ck.unlink()                                                                 # checkpoint disappears
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    day2 = (out / "2026-06-17.md").read_text(encoding="utf-8")
    assert "live base turn stays" in day2          # live content intact (not deleted)
    assert "recovered augment turn" not in day2    # stale recovered section dropped on forced rewrite


def test_openclaw_checkpoints_recursive_nested_store(tmp_path):
    # --recursive must reach NESTED checkpoints; dir-aware grouping resolves
    # the base in the checkpoint's OWN directory (not the top-level sessions dir).
    sd = tmp_path / "oc"
    nested = sd / "sub"
    append_jsonl(nested / "sess.jsonl", [_openclaw_session(),
                 _openclaw_msg("user", "nested base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(nested / "sess.checkpoint.aaaa.jsonl",
                   [("user", "nested deep turn long enough to keep", "2026-06-15T01:00:00Z")])
    # non-recursive (default): nested checkpoint is NOT reached
    out1 = tmp_path / "out1"
    extract_session.process_all(out_dir=out1, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert not (out1 / "2026-06-15.md").exists()
    # recursive: nested checkpoint recovered, base resolved in the nested dir
    out2 = tmp_path / "out2"
    extract_session.process_all(out_dir=out2, sessions_dir=sd, source="openclaw", recover_checkpoints=True, recursive=True)
    assert "nested deep turn" in (out2 / "2026-06-15.md").read_text(encoding="utf-8")
    assert "nested base turn" in (out2 / "2026-06-17.md").read_text(encoding="utf-8")


# --- Robustness: structurally-odd-but-valid-JSON input must not abort the run ---
# (Backs the README "Robust / Graceful on bad input" guarantee. Session logs are
# untrusted external data whose shape varies across tools/versions.)


def test_extract_text_blocks_tolerates_missing_and_null_text():
    # A content block typed "text" but with a missing/null/non-string text value must be
    # skipped, not raise KeyError/AttributeError.
    assert extract_session.extract_text_blocks([{"type": "text"}]) == ""
    assert extract_session.extract_text_blocks([{"type": "text", "text": None}]) == ""
    assert extract_session.extract_text_blocks([{"type": "text", "text": 123}]) == ""
    # A good block alongside a bad one still yields the good one.
    out = extract_session.extract_text_blocks(
        [{"type": "text", "text": None}, {"type": "text", "text": "keep me please long enough"}]
    )
    assert "keep me please" in out


def test_parsers_tolerate_non_dict_json_lines(tmp_path):
    # A top-level JSON line that is a bare array / string / number / null is valid JSON but not a
    # dict; every parser must skip it, not crash on obj.get(...).
    for src in ("claude", "codex", "openclaw"):
        p = tmp_path / f"{src}.jsonl"
        p.write_text('[]\n"a bare string"\n42\nnull\n', encoding="utf-8")
        assert extract_session.parse_jsonl(p, source=src) == []


def test_detect_source_tolerates_non_dict_lines(tmp_path):
    p = tmp_path / "weird.jsonl"
    p.write_text('[]\n"bare"\n42\n', encoding="utf-8")
    assert extract_session.detect_source(p) in extract_session.SOURCES  # must not raise


def test_one_bad_line_does_not_abort_the_sweep(tmp_path):
    # THE headline guarantee: a file with a structurally-odd line must not lose the OTHER good days
    # of the --all sweep. Unguarded, the null-text block aborts process_all before anything is written.
    sd = tmp_path / "sessions"
    out = tmp_path / "out"
    write_jsonl(sd / "good.jsonl", "user", "good content long enough to keep", "2026-06-10T01:00:00Z")
    bad = sd / "bad.jsonl"
    bad.write_text(
        json.dumps({"type": "user", "message": {"role": "user",
                    "content": [{"type": "text", "text": None}]}}) + "\n",
        encoding="utf-8",
    )
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="claude")  # must not raise
    assert (out / "2026-06-10.md").exists()  # the good day survived the bad sibling file
    assert "good content" in (out / "2026-06-10.md").read_text(encoding="utf-8")


def test_claude_project_filter_is_noop_not_empty(tmp_path):
    # --source claude --project X must NOT silently drop every session: Claude has no project
    # metadata, so --project is a no-op for it, not a total filter-out (session_project → None).
    sd = tmp_path / "sessions"
    out = tmp_path / "out"
    write_jsonl(sd / "s.jsonl", "user", "real claude content long enough to keep", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="claude", project="astro")
    assert (out / "2026-06-17.md").exists()  # not silently emptied
    assert "real claude content" in (out / "2026-06-17.md").read_text(encoding="utf-8")


def test_claude_project_does_not_poison_state_scope(tmp_path):
    # A Claude out-dir created WITHOUT --project must not reject an otherwise-identical rerun WITH
    # --project (and vice-versa): --project is normalized away for Claude before state scoping, so
    # it never lands in the state key.
    sd = tmp_path / "sessions"
    out = tmp_path / "out"
    write_jsonl(sd / "s.jsonl", "user", "claude content long enough to keep", "2026-06-17T01:00:00Z")
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="claude")  # no project
    # rerun WITH --project must not raise a scope mismatch
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="claude", project="astro")
    assert (out / "2026-06-17.md").exists()


def test_claude_list_with_project_is_not_emptied(tmp_path, capsys):
    # `--list --source claude --project X` must still show Claude rows (--project is inapplicable
    # to Claude and is normalized away, not used to filter every row out).
    sd = tmp_path / "sessions"
    write_jsonl(sd / "s.jsonl", "user", "claude content long enough to keep", "2026-06-17T01:00:00Z")
    extract_session.list_sessions(sessions_dir=sd, source="claude", project="astro")
    out = capsys.readouterr().out
    assert "2026-06-17" in out


def test_session_project_tolerates_non_dict_first_line(tmp_path):
    # The project probe must skip a non-object JSON line (e.g. a headerless/odd checkpoint) and
    # still read a later valid header, not crash on obj.get(...) — the project probe must not
    # reintroduce the bad-input crash.
    p = tmp_path / "rollout.jsonl"
    p.write_text(
        '[]\n' + json.dumps({"type": "session_meta", "payload": {"cwd": "/home/u/myrepo"}}) + "\n",
        encoding="utf-8",
    )
    assert extract_session.session_project(p, "codex") == "myrepo"  # must not raise


def test_infer_date_tolerates_odd_fractional_seconds(tmp_path):
    # 1/2/4/9-digit fractional seconds (which Python 3.9/3.10 fromisoformat rejects) must still
    # date correctly, not silently fall back to the file mtime.
    p = tmp_path / "s.jsonl"
    p.write_text("{}\n", encoding="utf-8")
    tz = extract_session.parse_timezone("+08:00")
    for frac in ("1", "12", "1934", "123456789"):
        ts = f"2026-06-26T20:00:00.{frac}Z"  # 20:00Z +08:00 → 04:00 next day → 2026-06-27
        assert extract_session.infer_date([{"ts": ts}], p, tz) == "2026-06-27"


def test_openclaw_checkpoints_window_warns_about_multiday_caveat(tmp_path, capsys):
    # --checkpoints + --since/--until can't perfectly reconcile a multi-day session (per-session
    # first-turn dating). Rather than silently drop OR duplicate its in-window turns, the tool WARNS.
    # (Seeding the dedup set unconditionally keeps the no-inflation contract; the warning makes the
    # windowed-multi-day limitation non-silent — the real fix is per-turn dating, --split-by-date.)
    sd = tmp_path / "oc"
    out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [
        _openclaw_session(),
        _openclaw_msg("user", "early base turn long enough to keep", "2026-06-10T01:00:00Z"),
        _openclaw_msg("user", "the in-window base turn long enough to keep", "2026-06-18T01:00:00Z"),
    ])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl",
                   [("user", "the in-window base turn long enough to keep", "2026-06-18T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw",
                                recover_checkpoints=True, since="2026-06-15", until="2026-06-20")
    err = capsys.readouterr().err.lower()
    assert "window" in err and "multi-day" in err  # the caveat is warned, not silent


def test_openclaw_checkpoints_no_window_recovers_without_warning(tmp_path, capsys):
    # The caveat warning fires ONLY with a window — an unwindowed --checkpoints run is unaffected.
    sd = tmp_path / "oc"
    out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl",
                 [_openclaw_session(), _openclaw_msg("user", "base turn long enough to keep", "2026-06-17T01:00:00Z")])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl",
                   [("user", "deep turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    err = capsys.readouterr().err.lower()
    assert "multi-day" not in err
    assert "deep turn" in (out / "2026-06-15.md").read_text(encoding="utf-8")


def test_openclaw_checkpoints_dedup_normalizes_ts_spelling(tmp_path):
    # The same instant spelled 'Z' in the base and '+00:00' in the checkpoint must dedup to ONE.
    sd = tmp_path / "oc"
    out = tmp_path / "out"
    append_jsonl(sd / "sess.jsonl", [
        _openclaw_session(),
        _openclaw_msg("user", "same instant turn long enough to keep", "2026-06-17T01:00:00Z"),
    ])
    _oc_checkpoint(sd / "sess.checkpoint.aaaa.jsonl",
                   [("user", "same instant turn long enough to keep", "2026-06-17T01:00:00+00:00")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw", recover_checkpoints=True)
    assert (out / "2026-06-17.md").read_text(encoding="utf-8").count("same instant turn") == 1


def test_openclaw_checkpoints_project_probe_scans_all_checkpoints(tmp_path):
    # base rotated away; one checkpoint is headerless. --project must resolve the parent's project
    # from ANY checkpoint that has a {type:session, cwd} header, not just an arbitrary first one.
    sd = tmp_path / "oc"
    out = tmp_path / "out"
    append_jsonl(sd / "sess.checkpoint.aaa.jsonl",  # headerless (sorts first)
                 [_openclaw_msg("user", "headerless deep turn long enough to keep", "2026-06-14T01:00:00Z")])
    append_jsonl(sd / "sess.checkpoint.bbb.jsonl",  # carries the project header (cwd → clawria-workspace)
                 [_openclaw_session(cwd="/home/u/clawria-workspace"),
                  _openclaw_msg("user", "headed deep turn long enough to keep", "2026-06-15T01:00:00Z")])
    extract_session.process_all(out_dir=out, sessions_dir=sd, source="openclaw",
                                recover_checkpoints=True, project="clawria-workspace")
    assert "headed deep turn" in (out / "2026-06-15.md").read_text(encoding="utf-8")
    assert "headerless deep turn" in (out / "2026-06-14.md").read_text(encoding="utf-8")