# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

"""Tests for segment orchestration in think."""

import importlib
import json
from pathlib import Path

import pytest
from jsonschema import Draft202012Validator

from solstone.observe.processing_record import (
    HANDLER_DESCRIBE,
    REASON_NO_DECODABLE_FRAMES,
    STATE_EMPTY,
    build_processing_record,
)


@pytest.fixture
def segment_dir(tmp_path, monkeypatch):
    """Create a temporary journal with a segment directory."""
    journal = tmp_path / "journal"
    day_dir = journal / "chronicle" / "20240115"
    segment_path = day_dir / "default" / "120000_300"
    segment_path.mkdir(parents=True)
    (segment_path / "talents").mkdir(parents=True)

    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
    return segment_path


def _segment_configs(*names: str) -> dict[str, dict]:
    configs = {
        "sense": {
            "priority": 10,
            "type": "generate",
            "output": "json",
            "schedule": "segment",
        },
        "documents": {
            "priority": 20,
            "type": "cogitate",
            "schedule": "segment",
        },
        "timeline:segment_summary": {
            "priority": 41,
            "type": "generate",
            "output": "json",
            "schedule": "segment",
        },
        "entities:detection": {
            "priority": 15,
            "type": "generate",
            "output": "json",
            "schedule": "segment",
        },
        "screen": {
            "priority": 20,
            "type": "generate",
            "output": "md",
            "schedule": "segment",
        },
        "speaker_attribution": {
            "priority": 20,
            "type": "cogitate",
            "schedule": "segment",
        },
        "pulse": {
            "priority": 30,
            "type": "cogitate",
            "schedule": "segment",
        },
    }
    return {name: dict(configs[name]) for name in names}


def _sense_config_with_load(*names: str) -> dict[str, dict]:
    cfgs = _segment_configs(*names)
    cfgs["sense"]["load"] = {
        "transcripts": True,
        "percepts": True,
        "talents": False,
    }
    return cfgs


def _read_jsonl_events(path: Path) -> list[dict]:
    return [
        json.loads(line)
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def _write_sense_output(segment_dir: Path, sense_json: dict) -> None:
    (segment_dir / "talents" / "sense.json").write_text(
        json.dumps(sense_json),
        encoding="utf-8",
    )


def _seed_empty_screen(segment_dir: Path, name: str = "screen") -> None:
    raw_path = segment_dir / f"{name}.webm"
    raw_path.write_bytes(b"raw-screen")
    header = {
        "raw": f"{name}.webm",
        "first_hash": None,
        "last_hash": None,
        "qualified_count": 0,
    }
    header["_solstone_processing"] = build_processing_record(
        state=STATE_EMPTY,
        reason_code=REASON_NO_DECODABLE_FRAMES,
        handler=HANDLER_DESCRIBE,
        input_size=raw_path.stat().st_size,
        source="backfill",
    )
    (segment_dir / f"{name}.jsonl").write_text(
        json.dumps(header) + "\n", encoding="utf-8"
    )


def _seed_browser_content(segment_dir: Path) -> None:
    rows = [
        {
            "t": "segment_start",
            "ts": 1,
            "site": "mail.google.com",
            "title": "Inbox - Gmail",
            "blocks": [
                {
                    "type": "row",
                    "text": (
                        "Browser-only sense input with enough semantic page text "
                        "to clear the no-input gate."
                    ),
                }
            ],
        },
        {
            "t": "delta",
            "ts": 2,
            "op": "add",
            "block": {
                "type": "row",
                "text": "Added browser update about the planning thread.",
            },
        },
    ]
    (segment_dir / "browser_mail-google-com.jsonl").write_text(
        "".join(json.dumps(row) + "\n" for row in rows),
        encoding="utf-8",
    )


def _active_sense_json() -> dict:
    return {
        "density": "active",
        "content_type": "coding",
        "activity_summary": "coding work",
        "recommend": {
            "screen_record": True,
            "speaker_attribution": True,
        },
        "facets": [{"facet": "work", "activity": "coding", "level": "high"}],
        "entities": [],
        "meeting_detected": False,
        "speakers": [],
    }


def _write_screen_header(
    segment_dir: Path,
    *,
    first_hash: str = "0000000000000000",
    last_hash: str = "0000000000000000",
) -> None:
    (segment_dir / "center_DP-3_screen.jsonl").write_text(
        json.dumps(
            {
                "raw": "center_DP-3_screen.webm",
                "first_hash": first_hash,
                "last_hash": last_hash,
                "qualified_count": 1,
            }
        )
        + "\n"
        + json.dumps({"timestamp": 1.0, "description": "active screen"})
        + "\n",
        encoding="utf-8",
    )


def _write_predecessor_change(
    predecessor_dir: Path,
    *,
    screen_last_hash: str = "0000000000000000",
    transcript: dict | None = None,
) -> None:
    if transcript is None:
        transcript = {
            "present": False,
            "word_count": 0,
            "content_hash": None,
        }
    payload = {
        "timestamp": "2024-01-15T11:55:00+00:00",
        "predecessor": None,
        "change_class": "active",
        "changed_sensors": ["screen"],
        "sensors": {
            "screen": {
                "monitors": {
                    "center:DP-3": {
                        "first_hash": "0000000000000000",
                        "last_hash": screen_last_hash,
                        "qualified_count": 1,
                    }
                }
            },
            "transcript": transcript,
        },
    }
    talents_dir = predecessor_dir / "talents"
    talents_dir.mkdir(parents=True, exist_ok=True)
    (talents_dir / "change.json").write_text(
        json.dumps(payload),
        encoding="utf-8",
    )


def _make_predecessor(segment_dir: Path) -> Path:
    predecessor_dir = segment_dir.parent / "115000_300"
    predecessor_dir.mkdir(parents=True, exist_ok=True)
    _write_predecessor_change(predecessor_dir)
    return predecessor_dir


def _patch_segment_cortex(monkeypatch, think, spawned: list[str]) -> None:
    monkeypatch.setattr(
        think,
        "get_talent_configs",
        lambda schedule=None, **kwargs: _segment_configs(
            "sense",
            "documents",
            "timeline:segment_summary",
            "screen",
            "speaker_attribution",
        ),
    )
    monkeypatch.setattr(
        think,
        "cortex_request",
        lambda prompt, name, config=None, **kwargs: (
            spawned.append(name) or f"agent-{name}"
        ),
    )
    monkeypatch.setattr(
        think,
        "wait_for_uses",
        lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
    )
    monkeypatch.setattr(think, "_callosum", None)


class TestLoadSegmentFacets:
    """Tests for load_segment_facets helper function."""

    def test_missing_file_returns_empty(self, segment_dir):
        from solstone.think.facets import load_segment_facets

        assert load_segment_facets("20240115", "120000_300") == []

    def test_empty_file_returns_empty(self, segment_dir):
        from solstone.think.facets import load_segment_facets

        (segment_dir / "talents" / "facets.json").write_text("")
        assert load_segment_facets("20240115", "120000_300") == []

    def test_empty_array_returns_empty(self, segment_dir):
        from solstone.think.facets import load_segment_facets

        (segment_dir / "talents" / "facets.json").write_text("[]")
        assert load_segment_facets("20240115", "120000_300") == []

    def test_valid_facets_extracted(self, segment_dir):
        from solstone.think.facets import load_segment_facets

        facets_data = [
            {"facet": "work", "activity": "Code review", "level": "high"},
            {"facet": "personal", "activity": "Email check", "level": "low"},
        ]
        (segment_dir / "talents" / "facets.json").write_text(json.dumps(facets_data))

        assert load_segment_facets("20240115", "120000_300") == ["work", "personal"]

    def test_malformed_json_returns_empty(self, segment_dir, caplog):
        from solstone.think.facets import load_segment_facets

        (segment_dir / "talents" / "facets.json").write_text("{ invalid json")
        assert load_segment_facets("20240115", "120000_300") == []
        assert "Failed to parse facets.json" in caplog.text

    def test_non_array_returns_empty(self, segment_dir, caplog):
        from solstone.think.facets import load_segment_facets

        (segment_dir / "talents" / "facets.json").write_text('{"facet": "work"}')
        assert load_segment_facets("20240115", "120000_300") == []
        assert "not an array" in caplog.text

    def test_missing_facet_field_skipped(self, segment_dir):
        from solstone.think.facets import load_segment_facets

        facets_data = [
            {"facet": "work", "activity": "Coding"},
            {"activity": "Unknown"},
            {"facet": "personal", "activity": "Email"},
        ]
        (segment_dir / "talents" / "facets.json").write_text(json.dumps(facets_data))

        assert load_segment_facets("20240115", "120000_300") == ["work", "personal"]


class TestRunSegmentSense:
    def test_recorded_empty_gates_via_data_state(self, segment_dir):
        from solstone.think.talents import (
            MIN_INPUT_CHARS,
            _is_no_input,
            _load_transcript,
            check_segment_has_no_input,
        )

        _seed_empty_screen(segment_dir)
        sources = {"transcripts": True, "percepts": True, "talents": False}

        transcript, counts = _load_transcript(
            "20240115", "120000_300", None, sources, stream="default"
        )

        assert _is_no_input(transcript, counts) is False
        assert len(transcript.strip()) >= MIN_INPUT_CHARS
        assert (
            check_segment_has_no_input(
                "20240115", "120000_300", sources, stream="default"
            )
            is True
        )

    def test_run_segment_sense_skips_raw_media_pending_before_idle_gate(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think
        from solstone.think.data_state import DataState
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_raw_pending.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "read_segment_data_state",
            lambda day, segment, stream=None: {"screen": DataState.PENDING.value},
        )
        monkeypatch.setattr(
            think,
            "check_segment_has_no_input",
            lambda *args, **kwargs: (_ for _ in ()).throw(
                AssertionError("idle gate should not run while raw media is pending")
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert result == (0, 0, [])
        assert spawned == []
        assert not (segment_dir / "talents" / "sense.json").exists()
        events = _read_jsonl_events(jsonl_path)
        skips = [event for event in events if event["event"] == "sense.skip"]
        assert skips == [
            {
                "event": "sense.skip",
                "ts": skips[0]["ts"],
                "mode": "segment",
                "day": "20240115",
                "segment": "120000_300",
                "reason": "raw_media_pending",
                "modalities": ["screen"],
                "stream": "default",
            }
        ]
        assert not any(event["event"] == "sense.complete" for event in events)

    def test_run_segment_sense_failed_only_falls_through_to_dispatch(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think
        from solstone.think.data_state import DataState
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_failed_state.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "conversation",
                "recommend": {},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense"),
        )
        monkeypatch.setattr(
            think,
            "read_segment_data_state",
            lambda day, segment, stream=None: {
                "audio": DataState.ANALYZED.value,
                "screen": DataState.FAILED.value,
            },
        )
        monkeypatch.setattr(think, "check_segment_has_no_input", lambda *a, **k: False)
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == ["sense"]
        assert (success, failed, failed_names) == (1, 0, [])
        events = _read_jsonl_events(jsonl_path)
        assert not any(event["event"] == "sense.skip" for event in events)
        assert any(event["event"] == "sense.complete" for event in events)

    def test_recorded_empty_segment_is_gated_to_idle(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.pipeline_health import (
            classify_segment_completion,
            read_segment_progress,
            segment_fully_thought,
        )
        from solstone.think.thinking import ThinkingJSONLWriter

        _seed_empty_screen(segment_dir)
        spawned = []
        jsonl_path = (
            segment_dir.parent.parent / "health" / "test_recorded_empty_gate.jsonl"
        )
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == []
        assert result == (0, 0, [])

        progress = read_segment_progress("20240115")
        prog = progress[("default", "120000_300")]
        assert segment_fully_thought(prog) == (True, None)

        segments = [
            {
                "key": "120000_300",
                "stream": "default",
                "data_state": {"screen": "empty"},
            }
        ]
        completion = classify_segment_completion(segments, progress)
        assert completion.not_thought == 0

        events = _read_jsonl_events(jsonl_path)
        sense_completes = [
            event for event in events if event["event"] == "sense.complete"
        ]
        assert len(sense_completes) == 1
        assert sense_completes[0]["density"] == "idle"
        assert sense_completes[0]["gated"] == "no_input"
        assert not any(event["event"] == "sense.skip" for event in events)
        assert any(event["event"] == "sense.change_detect" for event in events)
        assert not any(
            event["event"] in {"talent.complete", "talent.fail"}
            and event.get("name") == "sense"
            for event in events
        )

    def test_no_input_segment_is_gated_to_idle(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.pipeline_health import (
            classify_segment_completion,
            read_segment_progress,
            segment_fully_sensed,
            segment_fully_thought,
        )
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_gate.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == []
        assert result == (0, 0, [])

        progress = read_segment_progress("20240115")
        prog = progress[("default", "120000_300")]
        assert segment_fully_thought(prog) == (True, None)

        segments = [{"key": "120000_300", "stream": "default", "data_state": {}}]
        assert segment_fully_sensed(segments[0]["data_state"]) is True
        completion = classify_segment_completion(segments, progress)
        assert completion.not_sensed == 0
        assert completion.not_thought == 0

        events = _read_jsonl_events(jsonl_path)
        sense_completes = [
            event for event in events if event["event"] == "sense.complete"
        ]
        assert len(sense_completes) == 1
        assert sense_completes[0]["density"] == "idle"
        assert sense_completes[0]["gated"] == "no_input"
        assert any(event["event"] == "sense.change_detect" for event in events)
        assert not any(
            event["event"] in {"talent.complete", "talent.fail"}
            and event.get("name") == "sense"
            for event in events
        )
        assert not any(
            "output_parse" in failed_name
            for event in events
            for failed_name in event.get("failed_names", [])
        )

        sense_json = json.loads(
            (segment_dir / "talents" / "sense.json").read_text(encoding="utf-8")
        )
        schema_path = (
            Path(__file__).resolve().parents[1]
            / "solstone"
            / "talent"
            / "sense.schema.json"
        )
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
        errors = sorted(
            Draft202012Validator(schema).iter_errors(sense_json),
            key=lambda error: list(error.path),
        )
        assert not errors, [error.message for error in errors]
        assert "gated" not in sense_json

    def test_no_input_gate_fires_under_refresh(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.pipeline_health import (
            read_segment_progress,
            segment_fully_thought,
        )
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_gate_refresh.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=True,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == []
        assert result == (0, 0, [])
        progress = read_segment_progress("20240115")
        assert segment_fully_thought(progress[("default", "120000_300")]) == (
            True,
            None,
        )

    def test_no_input_gate_is_idempotent(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.pipeline_health import (
            read_segment_progress,
            segment_fully_thought,
        )
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_gate_repeat.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        first = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        second = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == []
        assert first == (0, 0, [])
        assert second == (0, 0, [])
        density = json.loads(
            (segment_dir / "talents" / "density.json").read_text(encoding="utf-8")
        )
        assert density["classification"] == "idle"
        progress = read_segment_progress("20240115")
        assert segment_fully_thought(progress[("default", "120000_300")]) == (
            True,
            None,
        )

    def test_recorded_empty_gate_is_idempotent(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.pipeline_health import (
            read_segment_progress,
            segment_fully_thought,
        )
        from solstone.think.thinking import ThinkingJSONLWriter

        _seed_empty_screen(segment_dir)
        spawned = []
        jsonl_path = (
            segment_dir.parent.parent / "health" / "test_recorded_empty_repeat.jsonl"
        )
        writer = ThinkingJSONLWriter(str(jsonl_path))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        first = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        second = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        assert spawned == []
        assert first == (0, 0, [])
        assert second == (0, 0, [])
        density = json.loads(
            (segment_dir / "talents" / "density.json").read_text(encoding="utf-8")
        )
        assert density["classification"] == "idle"
        progress = read_segment_progress("20240115")
        assert segment_fully_thought(progress[("default", "120000_300")]) == (
            True,
            None,
        )

    def test_mixed_empty_and_analyzed_dispatches_sense(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        _seed_empty_screen(segment_dir)
        (segment_dir / "audio.jsonl").write_text(
            "{}\n"
            '{"start":"00:00:01","text":"This transcript sentence is well over '
            'fifty characters so the gate must not fire."}\n',
            encoding="utf-8",
        )
        _write_sense_output(segment_dir, _active_sense_json())
        spawned = []

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense"]
        assert result == (1, 0, [])

    @pytest.mark.parametrize(
        ("filename", "body"),
        [
            (
                "audio.jsonl",
                "{}\n"
                '{"start":"00:00:01","text":"This transcript sentence is well '
                "over fifty characters so the no-input gate must not fire."
                '"}\n',
            ),
            (
                "screen.jsonl",
                '{"raw":"screen.webm"}\n'
                '{"timestamp":1,"analysis":{"primary":"work","visual_description":'
                '"This rendered screen description is well over fifty characters '
                'so the no-input gate must not fire."}}\n',
            ),
        ],
    )
    def test_input_present_dispatches_sense(
        self, segment_dir, monkeypatch, filename, body
    ):
        from solstone.think import thinking as think

        spawned = []
        (segment_dir / filename).write_text(body, encoding="utf-8")
        _write_sense_output(segment_dir, _active_sense_json())

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _sense_config_with_load("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        result = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense"]
        assert result == (1, 0, [])

    def test_browser_only_segment_is_not_gated_as_no_input(self, segment_dir):
        from solstone.think.cluster import _count_by_source, _load_entries_from_segment
        from solstone.think.talents import check_segment_has_no_input

        _seed_browser_content(segment_dir)
        sources = _sense_config_with_load("sense")["sense"]["load"]
        entries = _load_entries_from_segment(
            str(segment_dir),
            transcripts=False,
            percepts=True,
            agents=False,
        )
        counts = _count_by_source(entries)

        assert counts["percepts"] > 0
        assert (
            check_segment_has_no_input(
                "20240115",
                "120000_300",
                sources,
                stream="default",
            )
            is False
        )

    def test_check_segment_has_no_input_noop_without_sources(self, segment_dir):
        from solstone.think.talents import _is_no_input, check_segment_has_no_input

        assert segment_dir.is_dir()
        assert (
            check_segment_has_no_input("20240115", "120000_300", {}, stream="default")
            is False
        )
        assert _is_no_input("", {}) is True
        assert _is_no_input("x" * 49, {"transcripts": 1}) is True
        assert _is_no_input("x" * 60, {"transcripts": 1}) is False

    def test_sense_runs_first(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense", "documents"]
        assert success == 2
        assert failed == 0
        assert failed_names == []

    def test_idle_segment_returns_early(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        updates = []

        class StubStateMachine:
            def __init__(self):
                self.state = {}
                self.last_segment_key = None
                self.last_segment_day = None
                self.journal_root = segment_dir.parents[3]

            def update(self, sense_output, segment, day):
                updates.append((sense_output, segment, day))
                self.last_segment_key = segment
                self.last_segment_day = day
                return []

            def get_completed_activities(self):
                return []

        _write_sense_output(
            segment_dir,
            {
                "density": "idle",
                "content_type": "idle",
                "recommend": {"screen_record": True},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, _ = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            state_machine=StubStateMachine(),
        )

        assert spawned == ["sense"]
        assert success == 1
        assert failed == 0
        assert updates == [
            (
                {
                    "density": "idle",
                    "content_type": "idle",
                    "recommend": {"screen_record": True},
                    "facets": [],
                },
                "120000_300",
                "20240115",
            )
        ]
        density = json.loads((segment_dir / "talents" / "density.json").read_text())
        assert density["classification"] == "idle"

        # Verify activity state persisted even on idle path
        activity_state_path = (
            segment_dir.parents[3] / "awareness" / "activity_state.json"
        )
        assert activity_state_path.exists()
        state_data = json.loads(activity_state_path.read_text())
        assert state_data == {
            "last_segment_key": "120000_300",
            "last_segment_day": "20240115",
            "active": {},
        }

    def test_idle_short_circuit_persists_same_id_facet_records(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think
        from solstone.think.activities import load_activity_records
        from solstone.think.activity_state_machine import ActivityStateMachine

        spawned = []
        facets = [
            {"facet": "work", "activity": "deep parser work", "level": "high"},
            {"facet": "personal", "activity": "reading docs", "level": "medium"},
        ]
        entities = [
            {"type": "Person", "name": "Alice", "context": "pairing"},
            {"type": "Tool", "name": "VS Code", "context": "editor"},
        ]

        def sense(density="active", facet_payload=None):
            return {
                "density": density,
                "content_type": "coding",
                "activity_summary": "Working on parser state.",
                "entities": entities,
                "facets": facets if facet_payload is None else facet_payload,
                "meeting_detected": False,
                "speakers": [],
                "recommend": {"screen_record": True},
            }

        sm = ActivityStateMachine(journal_root=segment_dir.parents[3])
        sm.update(sense(), "115000_300", "20240115")
        sm.update(sense(), "115500_300", "20240115")
        _write_sense_output(segment_dir, sense(density="idle", facet_payload=[]))

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            state_machine=sm,
            skip_activity_prompts=True,
        )

        assert spawned == ["sense"]
        assert success == 1
        assert failed == 0
        assert failed_names == []

        work_records = load_activity_records("work", "20240115")
        personal_records = load_activity_records("personal", "20240115")
        assert len(work_records) == 1
        assert len(personal_records) == 1

        work = work_records[0]
        personal = personal_records[0]
        expected_segments = ["115000_300", "115500_300"]
        expected_entities = ["Alice", "VS Code"]
        assert work["description"] == "deep parser work"
        assert work["level_avg"] == 1.0
        assert work["segments"] == expected_segments
        assert work["active_entities"] == expected_entities
        assert personal["description"] == "reading docs"
        assert personal["level_avg"] == 0.5
        assert personal["segments"] == expected_segments
        assert personal["active_entities"] == expected_entities
        assert work["description"] != personal["description"]
        assert work["level_avg"] != personal["level_avg"]

    def test_conditional_screen_dispatch(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": True},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense", "documents", "screen"]

    @pytest.mark.parametrize("live", [False, True])
    def test_entities_detection_dispatches_in_stable_order(
        self, segment_dir, monkeypatch, live
    ):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense",
                "documents",
                "timeline:segment_summary",
                "entities:detection",
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            live=live,
        )

        assert spawned == [
            "sense",
            "documents",
            "timeline:segment_summary",
            "entities:detection",
        ]

    @pytest.mark.parametrize(
        ("has_embeddings", "expected"),
        [
            (False, ["sense", "documents"]),
            (True, ["sense", "documents", "speaker_attribution"]),
        ],
    )
    def test_conditional_speaker_attribution(
        self,
        segment_dir,
        monkeypatch,
        has_embeddings,
        expected,
    ):
        from solstone.think import thinking as think

        spawned = []
        if has_embeddings:
            (segment_dir / "audio.npz").write_bytes(b"npz")

        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"speaker_attribution": True},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense",
                "documents",
                "speaker_attribution",
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == expected

    def test_non_idle_dispatch_set_unchanged_with_change_detection(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think

        spawned = []
        (segment_dir / "audio.npz").write_bytes(b"npz")
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {
                    "screen_record": True,
                    "speaker_attribution": True,
                },
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense",
                "documents",
                "timeline:segment_summary",
                "screen",
                "speaker_attribution",
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == [
            "sense",
            "documents",
            "timeline:segment_summary",
            "screen",
            "speaker_attribution",
        ]

    def test_redundant_skips_writeups_and_writes_continuation(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think
        from solstone.think.change_detection import resolve_predecessor

        spawned = []
        _make_predecessor(segment_dir)
        _write_screen_header(segment_dir, first_hash="0000000000000000")
        (segment_dir / "audio.npz").write_bytes(b"npz")
        _write_sense_output(segment_dir, _active_sense_json())
        _patch_segment_cortex(monkeypatch, think, spawned)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            predecessor=resolve_predecessor("20240115", "default", "120000_300"),
        )

        assert (success, failed, failed_names) == (1, 0, [])
        assert spawned == ["sense"]
        change = json.loads((segment_dir / "talents" / "change.json").read_text())
        assert change["change_class"] == "redundant"
        assert change["sensors"]["transcript"]["present"] is False
        timeline = json.loads((segment_dir / "timeline.json").read_text())
        assert timeline["continuation_of"] == "115000_300"
        assert timeline["title"]
        assert timeline["description"]

    def test_active_changed_sensor_dispatches_full_set(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.change_detection import resolve_predecessor

        spawned = []
        _make_predecessor(segment_dir)
        _write_screen_header(segment_dir, first_hash="00000000000000ff")
        (segment_dir / "audio.npz").write_bytes(b"npz")
        _write_sense_output(segment_dir, _active_sense_json())
        _patch_segment_cortex(monkeypatch, think, spawned)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            predecessor=resolve_predecessor("20240115", "default", "120000_300"),
        )

        assert spawned == [
            "sense",
            "documents",
            "timeline:segment_summary",
            "screen",
            "speaker_attribution",
        ]
        change = json.loads((segment_dir / "talents" / "change.json").read_text())
        assert change["change_class"] == "active"
        assert change["changed_sensors"] == ["screen"]

    def test_changed_transcript_is_active_not_redundant(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.change_detection import resolve_predecessor

        spawned = []
        predecessor_dir = _make_predecessor(segment_dir)
        _write_predecessor_change(
            predecessor_dir,
            transcript={
                "present": True,
                "word_count": 1,
                "content_hash": "sha256:previous",
            },
        )
        _write_screen_header(segment_dir, first_hash="0000000000000000")
        (segment_dir / "audio.jsonl").write_text(
            json.dumps({})
            + "\n"
            + json.dumps(
                {
                    "start": "00:00:01",
                    "text": "one two three four five six seven eight",
                }
            )
            + "\n",
            encoding="utf-8",
        )
        (segment_dir / "audio.npz").write_bytes(b"npz")
        _write_sense_output(segment_dir, _active_sense_json())
        _patch_segment_cortex(monkeypatch, think, spawned)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            predecessor=resolve_predecessor("20240115", "default", "120000_300"),
        )

        assert spawned == [
            "sense",
            "documents",
            "timeline:segment_summary",
            "screen",
            "speaker_attribution",
        ]
        change = json.loads((segment_dir / "talents" / "change.json").read_text())
        assert change["change_class"] == "active"
        assert change["changed_sensors"] == ["transcript"]

    def test_refresh_bypasses_redundant(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.change_detection import resolve_predecessor

        spawned = []
        _make_predecessor(segment_dir)
        _write_screen_header(segment_dir, first_hash="0000000000000000")
        (segment_dir / "audio.npz").write_bytes(b"npz")
        _write_sense_output(segment_dir, _active_sense_json())
        _patch_segment_cortex(monkeypatch, think, spawned)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=True,
            verbose=False,
            stream="default",
            predecessor=resolve_predecessor("20240115", "default", "120000_300"),
        )

        assert spawned == [
            "sense",
            "documents",
            "timeline:segment_summary",
            "screen",
            "speaker_attribution",
        ]
        assert not (segment_dir / "timeline.json").exists()

    def test_redundant_continuation_is_idempotent(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.change_detection import resolve_predecessor

        spawned = []
        _make_predecessor(segment_dir)
        _write_screen_header(segment_dir, first_hash="0000000000000000")
        _write_sense_output(segment_dir, _active_sense_json())
        _patch_segment_cortex(monkeypatch, think, spawned)
        predecessor = resolve_predecessor("20240115", "default", "120000_300")

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            predecessor=predecessor,
        )
        first = (segment_dir / "timeline.json").read_bytes()
        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            predecessor=predecessor,
        )
        second = (segment_dir / "timeline.json").read_bytes()

        assert first == second

    def test_redundant_state_machine_matches_active(self, tmp_path, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.activity_state_machine import ActivityStateMachine
        from solstone.think.change_detection import resolve_predecessor

        def prepare(journal: Path) -> Path:
            seg_dir = journal / "chronicle" / "20240115" / "default" / "120000_300"
            seg_dir.mkdir(parents=True)
            (seg_dir / "talents").mkdir(parents=True)
            _make_predecessor(seg_dir)
            _write_screen_header(seg_dir, first_hash="0000000000000000")
            _write_sense_output(seg_dir, _active_sense_json())
            state = {
                "last_segment_key": "115000_300",
                "last_segment_day": "20240115",
                "active": {
                    "work": {
                        "id": "coding-115000_300",
                        "activity": "coding",
                        "state": "active",
                        "since": "115000_300",
                        "description": "coding work",
                        "level": "high",
                        "active_entities": [],
                        "facet": "work",
                        "segment": "115000_300",
                        "segments": ["115000_300"],
                    }
                },
            }
            state_path = journal / "awareness" / "activity_state.json"
            state_path.parent.mkdir(parents=True)
            state_path.write_text(json.dumps(state), encoding="utf-8")
            return seg_dir

        def run_case(
            journal: Path, *, refresh: bool
        ) -> tuple[ActivityStateMachine, bytes]:
            spawned: list[str] = []
            monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
            _patch_segment_cortex(monkeypatch, think, spawned)
            state_machine = ActivityStateMachine(journal_root=journal)
            think.run_segment_sense(
                "20240115",
                "120000_300",
                refresh=refresh,
                verbose=False,
                stream="default",
                state_machine=state_machine,
                predecessor=resolve_predecessor("20240115", "default", "120000_300"),
            )
            snapshot = (journal / "awareness" / "activity_state.json").read_bytes()
            return state_machine, snapshot

        redundant_journal = tmp_path / "redundant"
        active_journal = tmp_path / "active"
        prepare(redundant_journal)
        prepare(active_journal)

        redundant_sm, redundant_snapshot = run_case(redundant_journal, refresh=False)
        active_sm, active_snapshot = run_case(active_journal, refresh=True)

        assert redundant_snapshot == active_snapshot
        assert redundant_sm.state["work"]["segments"] == ["115000_300", "120000_300"]
        assert active_sm.state["work"]["segments"] == ["115000_300", "120000_300"]
        assert redundant_sm.state["work"]["_change"] == "continuing"
        assert active_sm.state["work"]["_change"] == "continuing"
        assert redundant_sm.get_completed_activities() == []
        assert active_sm.get_completed_activities() == []

    def test_refresh_bypasses_idle(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {"density": "idle", "content_type": "idle", "recommend": {}, "facets": []},
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=True,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense", "documents"]
        assert success == 2
        assert failed == 0
        assert failed_names == []

    def test_documents_always_runs(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": False},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "screen"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert "documents" in spawned
        assert "screen" not in spawned

    def test_capped_floor_talent_is_not_dispatched(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.thinking import ThinkingJSONLWriter

        spawned = []
        jsonl_path = segment_dir.parent.parent / "health" / "test_capped_floor.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": False},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "is_floor_talent_capped", lambda *args: True)
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        events = _read_jsonl_events(jsonl_path)
        skips = [event for event in events if event["event"] == "talent.skip"]

        assert spawned == ["sense"]
        assert any(
            skip["name"] == "documents" and skip["reason"] == "capped" for skip in skips
        )

    def test_refresh_bypasses_capped_floor_talent_gate(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": False},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "is_floor_talent_capped", lambda *args: True)
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=True,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense", "documents"]

    def test_capped_floor_talent_gate_does_not_skip_non_floor_talents(
        self, segment_dir, monkeypatch
    ):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": False},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: {
                **_segment_configs("sense", "timeline:segment_summary"),
                "flow": {
                    "priority": 20,
                    "type": "cogitate",
                    "schedule": "segment",
                },
            },
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "SEGMENT_FLOOR_TALENTS", ("flow",))
        monkeypatch.setattr(think, "is_floor_talent_capped", lambda *args: True)
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert "flow" not in spawned
        assert "timeline:segment_summary" in spawned

    def test_segment_summary_dispatched_for_non_idle(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {"screen_record": False},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "timeline:segment_summary"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert "timeline:segment_summary" in spawned

    def test_segment_summary_not_dispatched_on_idle(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {"density": "idle", "content_type": "idle", "recommend": {}, "facets": []},
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs(
                "sense", "documents", "timeline:segment_summary"
            ),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert "timeline:segment_summary" not in spawned

    def test_sense_failure_stops_orchestrator(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        spawned = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: (
                spawned.append(name) or f"agent-{name}"
            ),
        )

        def mock_wait_for_agents(agent_ids, timeout=600):
            return ({agent_ids[0]: "error"}, [])

        monkeypatch.setattr(think, "wait_for_uses", mock_wait_for_agents)
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert spawned == ["sense"]
        assert success == 0
        assert failed == 1
        assert failed_names == ["sense (error)"]

    def test_activity_state_machine_updated(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        updates = []
        activity_calls = []

        class StubStateMachine:
            def __init__(self):
                self.state = {}
                self.last_segment_key = None
                self.last_segment_day = None
                self.journal_root = segment_dir.parents[3]

            def update(self, sense_output, segment, day):
                updates.append((sense_output, segment, day))
                self.last_segment_key = segment
                self.last_segment_day = day
                self.state = {
                    "work": {
                        "facet": "work",
                        "state": "active",
                        "id": "coding_120000_300",
                    }
                }
                return [{"state": "ended", "id": "coding_120000_300", "facet": "work"}]

            def get_completed_activities(self):
                return [
                    {
                        "id": "coding_120000_300",
                        "facet": "work",
                        "activity": "coding",
                        "segments": ["120000_300"],
                        "level_avg": 0.5,
                        "description": "coding",
                        "active_entities": [],
                        "created_at": 1713200000000,
                    }
                ]

        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: f"agent-{name}",
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(
            think,
            "run_activity_prompts",
            lambda **kwargs: activity_calls.append(kwargs) or True,
        )
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
            state_machine=StubStateMachine(),
        )

        assert updates == [
            (
                {
                    "density": "active",
                    "content_type": "coding",
                    "recommend": {},
                    "facets": [],
                },
                "120000_300",
                "20240115",
            )
        ]
        assert activity_calls == [
            {
                "day": "20240115",
                "activity_id": "coding_120000_300",
                "facet": "work",
                "refresh": False,
                "verbose": False,
                "max_concurrency": 2,
            }
        ]
        activity_state_path = (
            segment_dir.parents[3] / "awareness" / "activity_state.json"
        )
        assert activity_state_path.exists()
        state_data = json.loads(activity_state_path.read_text())
        assert state_data == {
            "last_segment_key": "120000_300",
            "last_segment_day": "20240115",
            "active": {
                "work": {"facet": "work", "state": "active", "id": "coding_120000_300"}
            },
        }

    def test_generator_triggers_incremental_indexing(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        indexer_calls = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )
        (segment_dir / "talents" / "flow.md").write_text("flow", encoding="utf-8")

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: {
                **_segment_configs("sense"),
                "flow": {
                    "priority": 20,
                    "type": "generate",
                    "output": "md",
                    "schedule": "segment",
                },
            },
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: f"agent-{name}",
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(
            think,
            "run_queued_command",
            lambda cmd, day, timeout=60: indexer_calls.append(cmd) or True,
        )
        monkeypatch.setattr(think, "SEGMENT_FLOOR_TALENTS", ("flow",))
        monkeypatch.setattr(think, "_callosum", None)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert len(indexer_calls) == 1
        assert indexer_calls[0][:2] == ["journal", "indexer"]
        assert "--rescan-file" in indexer_calls[0]

    def test_send_failure_counted(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        calls = []
        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )

        def mock_cortex_request(prompt, name, config=None, **kwargs):
            calls.append(name)
            if name == "sense":
                return "agent-sense"
            raise think.CortexSpawnUnavailable()

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert calls[0] == "sense"
        assert calls[1:] == ["documents"]
        assert success == 1
        assert failed == 1
        assert failed_names == ["documents (send)"]

    def test_not_claimed_recorded_without_wait(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think

        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {},
                "facets": [],
            },
        )
        wait_calls = []
        jsonl_events = []
        emitted_events = []

        def mock_cortex_request(prompt, name, config=None, **kwargs):
            if name == "sense":
                return "agent-sense"
            raise think.CortexNotClaimed(use_id="lost-1")

        def fake_wait_for_uses(agent_ids, timeout=600):
            wait_calls.append(list(agent_ids))
            return {aid: "finish" for aid in agent_ids}, []

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense", "documents"),
        )
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)
        monkeypatch.setattr(think, "wait_for_uses", fake_wait_for_uses)
        monkeypatch.setattr(
            think,
            "_jsonl_log",
            lambda event, **fields: jsonl_events.append((event, fields)),
        )
        monkeypatch.setattr(
            think,
            "emit",
            lambda event, **fields: emitted_events.append((event, fields)),
        )
        monkeypatch.setattr(think, "_callosum", None)

        success, failed, failed_names = think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )

        assert success == 1
        assert failed == 1
        assert failed_names == ["documents (request_lost)"]
        assert all("lost-1" not in call for call in wait_calls)
        assert (
            "talent.fail",
            {
                "mode": "segment",
                "day": "20240115",
                "segment": "120000_300",
                "name": "documents",
                "use_id": "lost-1",
                "state": "request_lost",
                "provider": None,
                "model": None,
                "stream": "default",
            },
        ) in jsonl_events
        assert (
            "talent_completed",
            {
                "mode": "segment",
                "day": "20240115",
                "segment": "120000_300",
                "name": "documents",
                "use_id": "lost-1",
                "state": "request_lost",
            },
        ) in emitted_events


class TestDispatchCortexRequest:
    """Tests for _dispatch_cortex_request."""

    def test_returns_use_id_on_success(self, monkeypatch):
        from solstone.think import thinking as think

        calls = []

        def mock_cortex_request(**kwargs):
            calls.append(kwargs)
            return "agent-1"

        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 0)
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)

        result = think._dispatch_cortex_request(prompt="hi", name="test")

        assert result == "agent-1"
        assert len(calls) == 1

    def test_returns_none_when_spawn_unavailable(self, monkeypatch):
        from solstone.think import thinking as think

        def mock_cortex_request(**kwargs):
            raise think.CortexSpawnUnavailable(detail="FileNotFoundError")

        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 0)
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)

        result = think._dispatch_cortex_request(prompt="hi", name="test")

        assert result is None

    def test_returns_not_claimed_sentinel(self, monkeypatch):
        from solstone.think import thinking as think

        def mock_cortex_request(**kwargs):
            raise think.CortexNotClaimed(use_id="lost-1", detail="not claimed")

        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 0)
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)

        result = think._dispatch_cortex_request(prompt="hi", name="test")

        assert isinstance(result, think._NotClaimed)
        assert result.use_id == "lost-1"

    def test_dispatch_floor_zero_skips_local_provider_check(self, monkeypatch):
        from solstone.think import thinking as think

        def fail_local_needed():
            raise AssertionError("floor <= 0 must skip local-provider resolution")

        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 0)
        monkeypatch.setattr(think, "is_local_provider_needed", fail_local_needed)
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda **_kwargs: "agent-1",
        )

        assert think._dispatch_cortex_request(prompt="hi", name="test") == "agent-1"

    # This pair proves only orchestrated dispatch gates; direct cortex_request stays
    # interactive. Either test alone would be a weak assertion.
    def test_dispatch_cortex_request_gates_when_local_needed(
        self,
        monkeypatch,
    ):
        from solstone.think import thinking as think

        order = []
        think.reset_dispatch_admission_state()
        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 1)
        monkeypatch.setattr(think, "is_local_provider_needed", lambda: True)
        monkeypatch.setattr(think.admission, "read_available_bytes", lambda: 0)

        def spy_wait(stage, **_kwargs):
            order.append(("gate", stage))
            return 0.0

        def mock_cortex_request(**_kwargs):
            order.append(("request", "test"))
            return "agent-1"

        monkeypatch.setattr(think.admission, "wait_for_memory_headroom", spy_wait)
        monkeypatch.setattr(think, "cortex_request", mock_cortex_request)

        assert think._dispatch_cortex_request(prompt="hi", name="test") == "agent-1"
        assert order == [("gate", "think"), ("request", "test")]

    def test_direct_cortex_request_never_gates(self, tmp_path, monkeypatch):
        from solstone.think import cortex_client
        from solstone.think import thinking as think

        gate_calls = []
        sent = []
        monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
        think.reset_dispatch_admission_state()
        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 1)
        monkeypatch.setattr(think, "is_local_provider_needed", lambda: True)
        monkeypatch.setattr(think.admission, "read_available_bytes", lambda: 0)
        monkeypatch.setattr(
            think.admission,
            "wait_for_memory_headroom",
            lambda *args, **kwargs: gate_calls.append((args, kwargs)) or 0.0,
        )
        monkeypatch.setattr(
            cortex_client,
            "callosum_send_classified",
            lambda tract, event, **fields: sent.append((tract, event, fields)) or None,
        )
        monkeypatch.setattr(
            cortex_client, "get_use_log_status", lambda _use_id: "running"
        )

        result = cortex_client.cortex_request(
            prompt="hi",
            name="test",
            use_id="1713629000008",
        )

        assert result == "1713629000008"
        assert gate_calls == []
        assert sent[0][0:2] == ("cortex", "request")

    def test_dispatch_memory_throttle_emits_callosum_and_jsonl_waited_seconds(
        self,
        monkeypatch,
    ):
        from solstone.think import thinking as think

        emitted = []
        jsonl = []
        think.reset_dispatch_admission_state()
        monkeypatch.setattr(think.admission, "resolve_memory_floor_bytes", lambda: 1)
        monkeypatch.setattr(think, "is_local_provider_needed", lambda: True)
        monkeypatch.setattr(
            think, "emit", lambda event, **fields: emitted.append((event, fields))
        )
        monkeypatch.setattr(
            think,
            "_jsonl_log",
            lambda event, **fields: jsonl.append((event, fields)),
        )

        def fake_wait(stage, **kwargs):
            kwargs["on_throttle_start"](
                stage=stage,
                available_mib=2048,
                floor_mib=5120,
                available_bytes=2048 * 1024**2,
                floor_bytes=5120 * 1024**2,
            )
            kwargs["on_throttle_end"](stage=stage, waited_seconds=12.5)
            return 12.5

        monkeypatch.setattr(think.admission, "wait_for_memory_headroom", fake_wait)
        monkeypatch.setattr(think, "cortex_request", lambda **_kwargs: "agent-1")

        assert think._dispatch_cortex_request(prompt="hi", name="test") == "agent-1"
        assert emitted == [
            (
                "memory_throttle_started",
                {"stage": "think", "available_mib": 2048, "floor_mib": 5120},
            ),
            (
                "memory_throttle_completed",
                {"stage": "think", "waited_seconds": 12.5},
            ),
        ]
        assert jsonl == [
            (
                "memory_throttle.complete",
                {"stage": "think", "waited_seconds": 12.5},
            )
        ]


class TestStreamAutoResolution:
    """Tests for stream resolution in segment mode."""

    def test_auto_resolves_stream_from_filesystem(self, segment_dir, monkeypatch):
        mod = importlib.import_module("solstone.think.thinking")
        calls: list[dict] = []

        class MockCallosumConnection:
            def __init__(self, *args, **kwargs):
                pass

            def start(self, callback=None):
                return None

            def emit(self, *args, **kwargs):
                return None

            def stop(self):
                return None

        def mock_run_segment_sense(day, segment, refresh, verbose, **kwargs):
            calls.append(
                {
                    "day": day,
                    "segment": segment,
                    "refresh": refresh,
                    "verbose": verbose,
                    **kwargs,
                }
            )
            return (1, 0, [])

        monkeypatch.setattr(
            mod,
            "iter_segments",
            lambda day: [("mystream", "120000_300", Path("/tmp/segment"))],
        )
        monkeypatch.setattr(mod, "run_segment_sense", mock_run_segment_sense)
        monkeypatch.setattr(mod, "check_callosum_available", lambda: True)
        monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
        monkeypatch.setattr(
            mod, "run_queued_command", lambda cmd, day, timeout=600: True
        )
        monkeypatch.setattr(mod, "CallosumConnection", MockCallosumConnection)
        monkeypatch.setattr(
            "sys.argv",
            ["sol think", "--day", "20240115", "--segment", "120000_300"],
        )

        mod.main()

        assert len(calls) == 1
        assert calls[0]["stream"] == "mystream"

    def test_segment_not_found_exits(self, segment_dir, monkeypatch):
        mod = importlib.import_module("solstone.think.thinking")

        class MockCallosumConnection:
            def __init__(self, *args, **kwargs):
                pass

            def start(self, callback=None):
                return None

            def emit(self, *args, **kwargs):
                return None

            def stop(self):
                return None

        monkeypatch.setattr(mod, "iter_segments", lambda day: [])
        monkeypatch.setattr(
            mod, "run_segment_sense", lambda *args, **kwargs: (1, 0, [])
        )
        monkeypatch.setattr(mod, "check_callosum_available", lambda: True)
        monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
        monkeypatch.setattr(mod, "CallosumConnection", MockCallosumConnection)
        monkeypatch.setattr(
            "sys.argv",
            ["sol think", "--day", "20240115", "--segment", "999999_300"],
        )

        with pytest.raises(SystemExit) as excinfo:
            mod.main()

        assert excinfo.value.code != 0

    def test_explicit_stream_skips_filesystem_lookup(self, segment_dir, monkeypatch):
        mod = importlib.import_module("solstone.think.thinking")
        iter_calls = 0
        calls: list[dict] = []

        class MockCallosumConnection:
            def __init__(self, *args, **kwargs):
                pass

            def start(self, callback=None):
                return None

            def emit(self, *args, **kwargs):
                return None

            def stop(self):
                return None

        def mock_iter_segments(day):
            nonlocal iter_calls
            iter_calls += 1
            return [("mystream", "120000_300", Path("/tmp/segment"))]

        def mock_run_segment_sense(day, segment, refresh, verbose, **kwargs):
            calls.append(kwargs)
            return (1, 0, [])

        monkeypatch.setattr(mod, "iter_segments", mock_iter_segments)
        monkeypatch.setattr(mod, "run_segment_sense", mock_run_segment_sense)
        monkeypatch.setattr(mod, "check_callosum_available", lambda: True)
        monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
        monkeypatch.setattr(
            mod, "run_queued_command", lambda cmd, day, timeout=600: True
        )
        monkeypatch.setattr(mod, "CallosumConnection", MockCallosumConnection)
        monkeypatch.setattr(
            "sys.argv",
            [
                "sol think",
                "--day",
                "20240115",
                "--segment",
                "120000_300",
                "--stream",
                "explicit_stream",
            ],
        )

        mod.main()

        assert iter_calls == 0
        assert len(calls) == 1
        assert calls[0]["stream"] == "explicit_stream"


class TestThinkJSONLWriter:
    """Tests for ThinkingJSONLWriter."""

    def test_noop_when_no_path(self):
        from solstone.think.thinking import ThinkingJSONLWriter

        writer = ThinkingJSONLWriter(None)
        writer.log("test.event", foo="bar")
        writer.close()

        assert writer.skip_count == 0

    def test_writes_jsonl_to_file(self, tmp_path):
        from solstone.think.thinking import ThinkingJSONLWriter

        path = tmp_path / "test.jsonl"
        writer = ThinkingJSONLWriter(str(path))
        writer.log("run.start", mode="segment", day="20240115")
        writer.log(
            "talent.skip", name="screen", reason="not_recommended", detail="test"
        )
        writer.close()

        lines = path.read_text().strip().split("\n")
        assert len(lines) == 2

        first = json.loads(lines[0])
        assert first["event"] == "run.start"
        assert "ts" in first
        assert isinstance(first["ts"], int)
        assert first["mode"] == "segment"

        second = json.loads(lines[1])
        assert second["event"] == "talent.skip"
        assert writer.skip_count == 1

    def test_creates_parent_dirs(self, tmp_path):
        from solstone.think.thinking import ThinkingJSONLWriter

        path = tmp_path / "nested" / "dir" / "test.jsonl"
        writer = ThinkingJSONLWriter(str(path))
        writer.log("test.event")
        writer.close()

        assert path.exists()


class TestThinkJSONLEvents:
    """Tests for JSONL event emission during segment orchestration."""

    def test_density_idle_skip_event(self, segment_dir, monkeypatch):
        """JSONL emits talent.skip with reason=density_idle for idle segments."""
        from solstone.think import thinking as think
        from solstone.think.thinking import ThinkingJSONLWriter

        jsonl_path = segment_dir.parent.parent / "health" / "test_idle.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        _write_sense_output(
            segment_dir,
            {"density": "idle", "content_type": "idle", "recommend": {}, "facets": []},
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: "agent-sense",
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        events = [
            json.loads(line)
            for line in jsonl_path.read_text(encoding="utf-8").strip().splitlines()
        ]
        skips = [event for event in events if event["event"] == "talent.skip"]
        dispatches = [event for event in events if event["event"] == "talent.dispatch"]
        change_events = [
            event for event in events if event["event"] == "sense.change_detect"
        ]

        assert any(skip["reason"] == "density_idle" for skip in skips)
        assert [event["name"] for event in dispatches] == ["sense"]
        assert len(change_events) == 1
        assert change_events[0]["change_class"] == "idle"
        assert "changed_sensors" in change_events[0]

    def test_sense_complete_and_skip_events(self, segment_dir, monkeypatch):
        from solstone.think import thinking as think
        from solstone.think.thinking import ThinkingJSONLWriter

        jsonl_path = segment_dir.parent.parent / "health" / "test_think.jsonl"
        writer = ThinkingJSONLWriter(str(jsonl_path))

        _write_sense_output(
            segment_dir,
            {
                "density": "active",
                "content_type": "coding",
                "recommend": {
                    "screen_record": False,
                    "speaker_attribution": False,
                },
                "facets": [],
            },
        )

        monkeypatch.setattr(
            think,
            "get_talent_configs",
            lambda schedule=None, **kwargs: _segment_configs("sense"),
        )
        monkeypatch.setattr(
            think,
            "cortex_request",
            lambda prompt, name, config=None, **kwargs: f"agent-{name}",
        )
        monkeypatch.setattr(
            think,
            "wait_for_uses",
            lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
        )
        monkeypatch.setattr(think, "_callosum", None)
        monkeypatch.setattr(think, "_jsonl", writer)

        think.run_segment_sense(
            "20240115",
            "120000_300",
            refresh=False,
            verbose=False,
            stream="default",
        )
        writer.close()

        events = [
            json.loads(line)
            for line in jsonl_path.read_text(encoding="utf-8").strip().splitlines()
        ]
        assert "sense.complete" in [event["event"] for event in events]

        skips = [event for event in events if event["event"] == "talent.skip"]
        skip_pairs = {(event["name"], event["reason"]) for event in skips}
        assert ("documents", "no_config") in skip_pairs
        assert ("screen", "not_recommended") in skip_pairs
        assert ("speaker_attribution", "not_recommended") in skip_pairs
