agnes-the-ai-analyst/tests/test_capture_session.py

"""Tests for `agnes capture-session` — SessionStart hook helper."""

import json

from typer.testing import CliRunner

from cli.commands.capture_session import capture_session_app
from cli.lib.private_list import add_private
from cli.lib.session_queue import queue_path

runner = CliRunner()


def test_capture_appends_session_id_and_transcript_path(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    payload = json.dumps({
        "session_id": "abc-123",
        "transcript_path": "/Users/foo/.claude/projects/.../abc.jsonl",
        "cwd": "/Users/foo/work",
        "hook_event_name": "SessionStart",
    })
    result = runner.invoke(capture_session_app, [], input=payload)
    assert result.exit_code == 0
    assert queue_path(tmp_path).read_text(encoding="utf-8") == \
        "abc-123\t/Users/foo/.claude/projects/.../abc.jsonl\n"


def test_capture_writes_empty_session_id_when_field_missing(tmp_path, monkeypatch):
    """Payload missing session_id: still write to queue (legacy behavior).
    The empty session_id means the entry cannot be marked private later
    via /agnes-private (which keys on session_id), but it'll still upload."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    payload = json.dumps({"transcript_path": "/abc.jsonl"})
    result = runner.invoke(capture_session_app, [], input=payload)
    assert result.exit_code == 0
    assert queue_path(tmp_path).read_text(encoding="utf-8") == "\t/abc.jsonl\n"


def test_capture_silently_skips_when_field_missing(tmp_path, monkeypatch):
    """transcript_path is required; absence is a no-op."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    result = runner.invoke(capture_session_app, [], input='{"session_id": "abc"}')
    assert result.exit_code == 0
    assert not queue_path(tmp_path).exists()


def test_capture_silently_skips_invalid_json(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    result = runner.invoke(capture_session_app, [], input="not json {")
    assert result.exit_code == 0
    assert not queue_path(tmp_path).exists()


def test_capture_silently_skips_empty_stdin(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    result = runner.invoke(capture_session_app, [], input="")
    assert result.exit_code == 0
    assert not queue_path(tmp_path).exists()


def test_capture_silently_skips_non_string_transcript_path(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    payload = json.dumps({"transcript_path": 123})
    result = runner.invoke(capture_session_app, [], input=payload)
    assert result.exit_code == 0
    assert not queue_path(tmp_path).exists()


def test_capture_silently_skips_when_payload_not_object(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    result = runner.invoke(capture_session_app, [], input='["array", "not object"]')
    assert result.exit_code == 0
    assert not queue_path(tmp_path).exists()


def test_capture_appends_multiple_calls(tmp_path, monkeypatch):
    """Resume scenario: SessionStart fires again — same path appended twice."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    p1 = json.dumps({"session_id": "abc", "transcript_path": "/abc.jsonl"})
    p2 = json.dumps({"session_id": "abc", "transcript_path": "/abc.jsonl"})
    runner.invoke(capture_session_app, [], input=p1)
    runner.invoke(capture_session_app, [], input=p2)
    assert queue_path(tmp_path).read_text(encoding="utf-8") == (
        "abc\t/abc.jsonl\n"
        "abc\t/abc.jsonl\n"
    )


def test_capture_verbose_emits_diagnostic(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    payload = json.dumps({"session_id": "abc", "transcript_path": "/x.jsonl"})
    result = runner.invoke(capture_session_app, ["--verbose"], input=payload)
    assert result.exit_code == 0
    assert "/x.jsonl" in result.output


def test_capture_skips_queue_when_session_is_already_private(tmp_path, monkeypatch):
    """Race scenario: user ran /agnes-private BEFORE the SessionStart hook
    chain reached capture-session. The private list write happened first;
    capture-session sees it and skips the queue append."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    add_private(tmp_path, "abc-123")
    payload = json.dumps({
        "session_id": "abc-123",
        "transcript_path": "/abc.jsonl",
    })
    result = runner.invoke(capture_session_app, [], input=payload)
    assert result.exit_code == 0
    # Queue was NOT written.
    assert not queue_path(tmp_path).exists()


def test_capture_writes_when_unrelated_session_is_private(tmp_path, monkeypatch):
    """Marking session X private must not block capture of unrelated session Y."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    add_private(tmp_path, "other-session")
    payload = json.dumps({
        "session_id": "abc-123",
        "transcript_path": "/abc.jsonl",
    })
    result = runner.invoke(capture_session_app, [], input=payload)
    assert result.exit_code == 0
    assert queue_path(tmp_path).read_text(encoding="utf-8") == "abc-123\t/abc.jsonl\n"


# ---------- Breadcrumb tests (David #11 from PR review) ---------------------
#
# capture-session is a SessionStart hook that always exits 0 (it must NOT
# fail loudly inside Claude Code's startup chain). Without an external
# observability signal, an upstream contract change (Claude Code's stdin
# JSON shape shifts; the queue mysteriously stays empty) is invisible to
# operators. The breadcrumb log gives `agnes diagnose` something to
# inspect.

from cli.commands.capture_session import _BREADCRUMB_FILENAME


def _breadcrumb_lines(workspace) -> list[str]:
    """Read all breadcrumb lines, dropping trailing newline."""
    path = workspace / ".claude" / _BREADCRUMB_FILENAME
    if not path.exists():
        return []
    return [ln for ln in path.read_text(encoding="utf-8").splitlines() if ln]


def test_capture_writes_ok_breadcrumb_on_success(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    payload = json.dumps({"session_id": "abc-123", "transcript_path": "/abc.jsonl"})
    runner.invoke(capture_session_app, [], input=payload)
    lines = _breadcrumb_lines(tmp_path)
    assert len(lines) == 1
    parts = lines[0].split("\t")
    assert parts[1] == "ok"
    assert parts[2] == "abc-123"


def test_capture_writes_bad_json_breadcrumb_on_invalid_input(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    # Pre-create .claude/ so breadcrumb has somewhere to land (the breadcrumb
    # is read-only re: dir creation — see _record_breadcrumb).
    (tmp_path / ".claude").mkdir()
    runner.invoke(capture_session_app, [], input="not json at all")
    lines = _breadcrumb_lines(tmp_path)
    assert len(lines) == 1
    assert lines[0].split("\t")[1] == "bad_json"


def test_capture_writes_no_transcript_breadcrumb_when_field_missing(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    (tmp_path / ".claude").mkdir()
    runner.invoke(capture_session_app, [], input=json.dumps({"session_id": "x"}))
    lines = _breadcrumb_lines(tmp_path)
    assert len(lines) == 1
    assert lines[0].split("\t")[1] == "no_transcript_path"


def test_capture_writes_private_skip_breadcrumb_on_marked_session(tmp_path, monkeypatch):
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    add_private(tmp_path, "private-sid")
    runner.invoke(
        capture_session_app, [],
        input=json.dumps({"session_id": "private-sid", "transcript_path": "/x"}),
    )
    lines = _breadcrumb_lines(tmp_path)
    assert lines[-1].split("\t")[1] == "private_skip"
    assert lines[-1].split("\t")[2] == "private-sid"


def test_breadcrumb_does_not_create_claude_dir_in_arbitrary_workspaces(tmp_path, monkeypatch):
    """If the workspace has no .claude/ directory, capture-session
    never materializes one — same rationale as the read-only path in
    private_list.py. Hooks fire in directories the user just opened
    Claude Code in; the breadcrumb log shouldn't pollute those."""
    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
    runner.invoke(capture_session_app, [], input="malformed")
    # No .claude/ created, no breadcrumb written.
    assert not (tmp_path / ".claude").exists()