#!/usr/bin/env python3
"""Tests for the session record and the local stream (FLY-1411).

Run: python3 test_session_stream.py

Covers `session.py`'s half of phase 10: the SessionUpdate v1 envelope, the
relay write that replaced the direct project-update post, and `stream.jsonl` —
the append-only local record that is the whole feature on the local tier and
the offline copy on the cloud one.

Hermetic in the same sense as `test_enforcement.py`: no test here reaches the
network. Nothing constructs a real client — every case injects a fake through
`session.get_client`, and the one case that shells out runs `graph_session.py`
against a temporary project root.
"""

import contextlib as _contextlib
import datetime as _dt
import io as _io
import json
import os
import sys
import tempfile
import types as _types
from pathlib import Path
from unittest.mock import patch

SCRIPT_DIR = Path(__file__).parent.resolve()
sys.path.insert(0, str(SCRIPT_DIR))

# Set before flydocs_api is imported so no credential resolution can find a
# real key, in the unlikely event a future test builds a real client.
os.environ["FLYDOCS_API_KEY"] = "fdk_test_session_stream_hermetic"
os.environ["FLYDOCS_RELAY_URL"] = "http://127.0.0.1:9/api/relay"

import session  # noqa: E402
from flydocs_api import RelayError  # noqa: E402

passed = 0
failed = 0


def test(name: str):
    """Decorator to register and run a test."""
    def decorator(fn):
        global passed, failed
        try:
            fn()
            print(f"  PASS: {name}")
            passed += 1
        except AssertionError as e:
            print(f"  FAIL: {name} — {e}")
            failed += 1
        except Exception as e:
            print(f"  ERROR: {name} — {type(e).__name__}: {e}")
            failed += 1
        return fn
    return decorator


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

WORKSPACE_ID = "ws_stream_tests"

GOOD_BODY = (
    "## Accomplished\n\nThe record landed.\n\n"
    "## Next up\n\nDistribution.\n\n"
    "## Blockers & open questions\n\nNone.\n\n"
    "## Progress\n\n1 issue closed (FLY-1411).\n"
)


class _FakeCloudClient:
    """Cloud client whose `/session-updates` write plays a scripted outcome."""

    def __init__(self, outcome=None, config=None, fallback=None):
        self.tier = "cloud"
        self.is_cloud = True
        self.config = config if config is not None else {"activeProjectId": "proj_1"}
        self.outcome = outcome if outcome is not None else {
            "success": True,
            "id": "rec_1",
            "permalinkPath": "/acme/activity/rec_1",
            "replayed": False,
            "providerUpdate": {"posted": True, "id": "upd_1"},
        }
        # Only the fallback post reaches `project_update` on this tier; every
        # test asserts whether it was called, so an accidental direct post is
        # still caught.
        self.fallback = fallback if fallback is not None else {
            "success": True, "id": "upd_fallback",
        }
        self.calls = []
        self.fallback_calls = []

    def session_update_create(self, envelope, project_id=None):
        self.calls.append({"envelope": envelope, "projectId": project_id})
        if isinstance(self.outcome, Exception):
            raise self.outcome
        return self.outcome

    def project_update(self, health, body, project_id=None):
        self.fallback_calls.append(
            {"health": health, "body": body, "projectId": project_id})
        if isinstance(self.fallback, Exception):
            raise self.fallback
        return self.fallback


class _FakeLocalClient:
    """Local tier: no relay, the file-store update, and the stream."""

    def __init__(self):
        self.tier = "local"
        self.is_cloud = False
        self.config = {}
        self.calls = []

    def project_update(self, health, body, project_id=None):
        self.calls.append({"health": health, "body": body, "projectId": project_id})
        return {"success": True, "id": "local-update"}

    def session_update_create(self, *_a, **_k):  # pragma: no cover - guard
        raise AssertionError("the local tier has no relay to write a record to")


class _Response:
    """The minimal shape `_request` reads out of `urlopen`."""

    def __init__(self, body, status=201):
        self.status = status
        self.headers = {}
        self._body = json.dumps(body).encode()

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, *_exc):
        return False


def _real_cloud_client(root, base_url="http://127.0.0.1:9/api/relay"):
    """A genuine `FlyDocsClient` on the cloud tier, pointed nowhere real.

    The fakes above cannot answer the question these tests ask — whether the
    HTTP layer *exits* or *raises* — because they never enter it. 127.0.0.1:9
    is the discard port: the connection is refused immediately, so the
    unreachable-relay path runs at full speed.
    """
    import flydocs_api
    with patch.object(flydocs_api, "find_project_root", lambda: root):
        client = flydocs_api.FlyDocsClient()
    client._relay.base_url = base_url
    # One attempt: the retry loop's sleeps are not what is under test.
    client._relay.MAX_RETRIES = 1
    return client


def _project(session_files=None):
    """A temporary project root with a config and optional session files."""
    root = Path(tempfile.mkdtemp(prefix="flydocs-stream-"))
    (root / ".flydocs").mkdir()
    (root / ".flydocs" / "config.json").write_text(
        json.dumps({"workspaceId": WORKSPACE_ID, "repoSlug": "acme/widgets",
                    "tier": "cloud"})
    )
    session_dir = root / ".flydocs" / "session" / WORKSPACE_ID
    session_dir.mkdir(parents=True)
    for name, content in (session_files or {}).items():
        (session_dir / name).write_text(content)
    return root, session_dir


def _wrap(root, client, **flags):
    """Run `session.py wrap` in-process. Returns (exit_code, payload, stderr)."""
    args = _types.SimpleNamespace(
        issues=["FLY-1411"], health="onTrack", notes="", title="",
        pending=["ship it"], blockers=[], body=GOOD_BODY, body_file=None,
        project=None, visibility=None, started_at=None,
    )
    for key, value in flags.items():
        setattr(args, key, value)
    out, err = _io.StringIO(), _io.StringIO()
    code = 0
    with patch.object(session, "get_client", lambda: client), \
            patch.object(session, "find_project_root", lambda: root), \
            _contextlib.redirect_stdout(out), _contextlib.redirect_stderr(err):
        try:
            session.cmd_wrap(args)
        except SystemExit as exit_error:
            code = exit_error.code or 0
    raw = out.getvalue()
    payload = json.loads(raw) if raw.strip() else None
    return code, payload, err.getvalue()


def _stream_lines(session_dir):
    path = session_dir / "stream.jsonl"
    if not path.exists():
        return []
    return [json.loads(line) for line in path.read_text().splitlines() if line]


def _action(payload, name):
    for entry in payload["actions"]:
        if entry.get("action") == name:
            return entry
    return None


# ---------------------------------------------------------------------------
# The wrap template resource
# ---------------------------------------------------------------------------

print("\n## wrap template resource")


@test("the template resource is named, versioned, and is the section list")
def _():
    assert session.WRAP_TEMPLATE["id"] == "session-wrap"
    assert session.WRAP_TEMPLATE["version"] == 1
    assert session.REQUIRED_WRAP_SECTIONS is \
        session.WRAP_TEMPLATE["required_sections"], \
        "the alias must BE the field, not a copy of it"


@test("every required section is a heading in the shipped wrap template")
def _():
    # The relay validates the same list server-side. If the template a caller
    # is told to fill stopped containing one of them, both validators would
    # reject every honest wrap — so the template is the third party that has
    # to agree.
    template = (SCRIPT_DIR.parent / "templates" / "session"
                / "session-wrap.md").read_text()
    headings = [
        line.lstrip("#").strip().lower()
        for line in template.splitlines() if line.startswith("#")
    ]
    for section in session.REQUIRED_WRAP_SECTIONS:
        assert any(section.lower() in heading for heading in headings), \
            f"no heading in session-wrap.md contains {section!r}"


# ---------------------------------------------------------------------------
# Envelope composition
# ---------------------------------------------------------------------------

print("\n## envelope composition")


@test("compose_session_update builds the v1 client envelope")
def _():
    envelope = session.compose_session_update(
        repo="acme/widgets", session_id="2026-08-29", actor={"agent": "claude-code"},
        started_at=1000, ended_at=2000, health="onTrack",
        issues=[{"ref": "FLY-1"}], pending=["next"], blockers=[{"text": "none"}],
        narrative=GOOD_BODY, visibility="team", cli_version="1.3.13",
    )
    assert envelope["schemaVersion"] == 1
    assert envelope["repo"] == "acme/widgets"
    assert envelope["sessionId"] == "2026-08-29"
    assert envelope["window"] == {"startedAt": 1000, "endedAt": 2000}
    assert envelope["narrative"]["markdown"] == GOOD_BODY, \
        "the narrative is the posted body byte for byte"
    assert envelope["shipped"] == [] and envelope["decisions"] == []
    assert "metrics" not in envelope, "v1 writes no metrics"
    assert envelope["provenance"] == {"source": "cli", "cliVersion": "1.3.13"}


@test("an unknown CLI version omits the field rather than inventing one")
def _():
    envelope = session.compose_session_update(
        repo="r", session_id="s", actor={}, started_at=1, ended_at=1,
        health="atRisk", issues=[], pending=[], blockers=[], narrative="x",
        visibility="private",
    )
    assert envelope["provenance"] == {"source": "cli"}


@test("the session id always carries a sequence, taken from graph and stream")
def _():
    root, session_dir = _project()
    graph_dir = root / "flydocs" / "context"
    graph_dir.mkdir(parents=True)

    assert session.resolve_session_id(root, session_dir, "2026-08-29") == \
        "2026-08-29-1", "the first session of a day is -1, never bare"

    # A wrap that never reached the graph (no notes, the MCP default) still
    # advances the sequence, because the stream is the other source.
    (session_dir / "stream.jsonl").write_text(
        json.dumps({"sessionId": "2026-08-29-1"}) + "\n"
        + json.dumps({"sessionId": "2026-08-29-2"}) + "\n"
        + json.dumps({"sessionId": "2026-08-28-9"}) + "\n"
    )
    assert session.resolve_session_id(root, session_dir, "2026-08-29") == \
        "2026-08-29-3", "yesterday's sequence does not count toward today's"

    # And a node from before the scheme had sequences is still not re-used.
    (graph_dir / "graph.json").write_text(json.dumps({
        "nodes": {"session:2026-08-29": {"type": "session"},
                  "session:2026-08-29-7": {"type": "session"}},
        "edges": [],
    }))
    assert session.resolve_session_id(root, session_dir, "2026-08-29") == \
        "2026-08-29-8"


@test("a rotated generation still counts toward today's sequence")
def _():
    root, session_dir = _project()
    (session_dir / "stream.1.jsonl").write_text(
        json.dumps({"sessionId": "2026-08-29-4"}) + "\n")
    assert session.resolve_session_id(root, session_dir, "2026-08-29") == \
        "2026-08-29-5"


@test("the graph summary falls back to the narrative when notes are empty")
def _():
    assert session.narrative_summary(GOOD_BODY) == \
        "Accomplished — The record landed."
    assert session.narrative_summary("no headings here") is None
    assert session.narrative_summary("") is None


@test("--started-at reads ISO and epoch ms; anything else falls back")
def _():
    assert session._parse_started_at("1756400000000") == 1756400000000
    iso = _dt.datetime(2026, 8, 29, tzinfo=_dt.timezone.utc)
    assert session._parse_started_at("2026-08-29T00:00:00Z") == \
        int(iso.timestamp() * 1000)
    assert session._parse_started_at("not a time") is None
    assert session._parse_started_at(None) is None


# ---------------------------------------------------------------------------
# Local tier
# ---------------------------------------------------------------------------

print("\n## local tier")


@test("a local wrap appends the record, posts locally, and needs no relay")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    client = _FakeLocalClient()
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert payload["success"] is True, payload
    lines = _stream_lines(session_dir)
    assert len(lines) == 1, lines
    record = lines[0]
    assert record["narrative"]["markdown"] == GOOD_BODY
    assert record["visibility"] == "team"
    assert record["local"]["relay"] == {"state": "skipped"}, record["local"]
    assert record["local"]["windowSource"] == "endedAt"
    assert record["window"]["startedAt"] == record["window"]["endedAt"]
    assert client.calls and client.calls[0]["health"] == "onTrack"
    assert _action(payload, "session_stream")["relayState"] == "skipped"


@test("wrapping without health writes no record at all")
def _():
    # No health means no update is posted — the documented escape hatch. A
    # record needs a health, so the escape hatch produces none rather than one
    # with a field invented for it.
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    code, payload, err = _wrap(root, _FakeLocalClient(), health=None)
    assert code == 0, err
    assert _stream_lines(session_dir) == []
    assert _action(payload, "session_stream") is None
    assert not (session_dir / "focus.md").exists(), "state is still cleared"


@test("statusTo comes from the status pair the session tracked")
def _():
    root, session_dir = _project({
        "focus.md": "FLY-1411\n", "status": "IMPLEMENTING",
        "status-ref": "FLY-1411",
    })
    _wrap(root, _FakeLocalClient(), issues=["FLY-1411", "FLY-9"])
    issues = _stream_lines(session_dir)[0]["issues"]
    by_ref = {entry["ref"]: entry for entry in issues}
    assert by_ref["FLY-1411"]["statusTo"] == "IMPLEMENTING", issues
    assert "statusTo" not in by_ref["FLY-9"], \
        "the session tracked one ref; the others get no status claim"


@test("--visibility and --started-at reach the record")
def _():
    root, session_dir = _project()
    _wrap(root, _FakeLocalClient(),
          visibility="private", started_at="1756400000000")
    record = _stream_lines(session_dir)[0]
    assert record["visibility"] == "private"
    assert record["window"]["startedAt"] == 1756400000000
    assert record["window"]["endedAt"] >= record["window"]["startedAt"]
    assert record["local"]["windowSource"] == "startedAt-flag"


# ---------------------------------------------------------------------------
# FLY-1498 — the title and the summary
#
# The two fields a human writes, and the two every destination opens with. The
# bounds under test are the *relay's*, and the relay counts UTF-16 code units
# (JavaScript `String.length`) — so an emoji-heavy summary that Python calls
# 400 characters is 800 to the validator, and the cases below say so.
# ---------------------------------------------------------------------------

print("\n## title and summary")

TITLE = "Landed the Activity week view over live data"
NOTES = (
    "The week view reads the session records instead of the fixture, so a "
    "wrap shows up on the timeline within the minute. The Slack renderer is "
    "the last consumer still on derived text."
)


@test("--title and --notes reach the request body and the stream line")
def _():
    root, session_dir = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title=TITLE, notes=NOTES)

    sent = client.calls[0]["envelope"]
    assert sent["title"] == TITLE, sent.get("title")
    assert sent["summary"] == NOTES, sent.get("summary")
    # The stream is the same bytes, which is the whole point of the local copy.
    record = _stream_lines(session_dir)[0]
    assert record["title"] == TITLE and record["summary"] == NOTES


@test("--notes still labels the graph node while it is the summary")
def _():
    # The field gained a destination; it did not lose one.
    root, _ = _project()
    calls = []
    with patch.object(session.subprocess, "run",
                      lambda cmd, **kw: calls.append(cmd) or _types.SimpleNamespace(
                          stdout=b"{}", returncode=0)):
        _wrap(root, _FakeCloudClient(), notes=NOTES)
    assert calls, "the graph script was never called"
    cmd = calls[0]
    assert NOTES in cmd, "the notes are still the graph node's summary"


@test("a wrap with neither sends neither key, and is the body it always was")
def _():
    root, session_dir = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title="   ", notes="   ")

    sent = client.calls[0]["envelope"]
    assert "title" not in sent and "summary" not in sent, \
        "blank is omission — the mutation refuses an empty string and a null"
    assert "title" not in _stream_lines(session_dir)[0]


@test("an oversize title and summary are cut at the relay's bounds")
def _():
    root, _ = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title="T" * 200, notes="Z" * 900)

    sent = client.calls[0]["envelope"]
    assert session._utf16_len(sent["title"]) == session.SESSION_TITLE_MAX
    assert session._utf16_len(sent["summary"]) == session.SESSION_SUMMARY_MAX


@test("an emoji-heavy pair is cut by UTF-16 units, not by code points")
def _():
    # Python's len() counts one; the validator counts two. Counting Python's
    # would send a 240-unit title to a route whose limit is 120.
    root, _ = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title="\U0001f600" * 200, notes="\U0001f600" * 900)

    sent = client.calls[0]["envelope"]
    assert session._utf16_len(sent["title"]) <= session.SESSION_TITLE_MAX
    assert session._utf16_len(sent["summary"]) <= session.SESSION_SUMMARY_MAX
    assert len(sent["title"]) == session.SESSION_TITLE_MAX // 2, \
        "an astral character is dropped whole, never split into surrogates"
    assert sent["title"].encode("utf-8").decode("utf-8") == sent["title"]


@test("a summary long enough to cut ends on a sentence, not mid-word")
def _():
    root, _ = _project()
    client = _FakeCloudClient()
    _wrap(root, client, notes="The view reads live data. " * 40)

    summary = client.calls[0]["envelope"]["summary"]
    assert summary.endswith("."), summary[-40:]
    assert session._utf16_len(summary) <= session.SESSION_SUMMARY_MAX


@test("a title cut on a word boundary carries no trailing space")
def _():
    # The route trims what it stores. Without the rstrip the record would hold
    # one title and the wrap would echo back another, differing by a space —
    # the sort of one-character disagreement nobody thinks to look for.
    word = "alpha "                       # 6 units
    title = word * 20 + "omega"           # unit 120 is the space after #20
    root, _ = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title=title)

    sent = client.calls[0]["envelope"]["title"]
    assert sent == sent.rstrip(), repr(sent[-8:])
    assert sent.endswith("alpha"), repr(sent[-12:])
    assert session._utf16_len(sent) == session.SESSION_TITLE_MAX - 1


@test("a line break in a title becomes a space rather than a 400")
def _():
    # A title read out of a file arrives with CRLF; the route refuses any
    # break, so the collapse happens here instead of over HTTP.
    root, _ = _project()
    client = _FakeCloudClient()
    _wrap(root, client, title="Landed the week view\r\n over\tlive   data\u2028now")

    title = client.calls[0]["envelope"]["title"]
    assert title == "Landed the week view over live data now", repr(title)


@test("the title is hashed into the operation id — a new title is a new intent")
def _():
    import flydocs_api
    root, _ = _project()
    client = _real_cloud_client(root)
    seen = []

    def fake_urlopen(request, timeout=None):
        headers = {k.lower(): v for k, v in request.headers.items()}
        seen.append(headers.get("x-operation-id"))
        return _Response({"success": True, "id": "rec"})

    def envelope(title):
        return session.compose_session_update(
            repo="acme/widgets", session_id="2026-08-30-1", actor={},
            started_at=1, ended_at=1, health="onTrack", issues=[], pending=[],
            blockers=[], narrative="one", visibility="team", title=title,
        )

    flydocs_api.set_operation_seed("op-seed-title")
    try:
        with patch("urllib.request.urlopen", fake_urlopen):
            client.session_update_create(envelope("First"))
            client.session_update_create(envelope("First"))
            client.session_update_create(envelope("Second"))
    finally:
        flydocs_api.set_operation_seed(None)

    assert seen[0] == seen[1], "the same wrap keys the same way"
    assert seen[2] != seen[0], "a different title is a different intent"


@test("the route's own words on a bounds rejection reach the caller")
def _():
    root, session_dir = _project({"focus.md": "FLY-1498\n"})
    error = RelayError(400, "ENVELOPE_INVALID", "title is 140 characters — the limit is 120",
                       {"code": "ENVELOPE_INVALID",
                        "error": "title is 140 characters — the limit is 120",
                        "findings": [{"field": "title", "message": "too long"}]})
    code, _payload, err = _wrap(root, _FakeCloudClient(outcome=error),
                                title="T" * 140)

    assert code == 1, err
    assert "title is 140 characters — the limit is 120" in err, err
    assert "title (too long)" in err, err
    assert _stream_lines(session_dir) == [], "a refused record is not a record"
    assert (session_dir / "focus.md").exists()


@test("the wrap echoes the pair it actually sent")
def _():
    root, _ = _project()
    _code, payload, _err = _wrap(root, _FakeCloudClient(),
                                 title="A" * 200, notes=NOTES)
    stored = _action(payload, "session_update")
    assert session._utf16_len(stored["title"]) == session.SESSION_TITLE_MAX, \
        "the caller sees the normalized value, not the one it wrote"
    assert stored["summary"] == NOTES

    _code, payload, _err = _wrap(root, _FakeCloudClient())
    stored = _action(payload, "session_update")
    assert "title" not in stored and "summary" not in stored


# ---------------------------------------------------------------------------
# FLY-990 — a rejected body changes nothing
# ---------------------------------------------------------------------------

print("\n## rejected bodies")


@test("a body missing sections appends nothing, posts nothing, cleans nothing")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n", "status": "IMPLEMENTING"})
    client = _FakeLocalClient()
    code, payload, err = _wrap(root, client, body="Did some things today.")

    assert code == 1, f"expected the guard to refuse: {payload}"
    assert client.calls == [], "nothing may be posted"
    assert _stream_lines(session_dir) == [], "and nothing may be appended"
    assert (session_dir / "focus.md").exists()
    assert (session_dir / "status").exists()
    assert "missing required section" in err


@test("a relay wrap rejection is the same refusal, findings and all")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    error = RelayError(400, "WRAP_VALIDATION_FAILED",
                       "narrative is missing required sections",
                       {"code": "WRAP_VALIDATION_FAILED",
                        "error": "narrative is missing required sections",
                        "findings": [{"section": "Blockers", "status": "missing"}]})
    client = _FakeCloudClient(outcome=error)
    code, _payload, err = _wrap(root, client)

    assert code == 1, err
    assert _stream_lines(session_dir) == [], \
        "a rejected body is not a record, wherever the rejection came from"
    assert (session_dir / "focus.md").exists()
    assert "Blockers (missing)" in err, err


# ---------------------------------------------------------------------------
# Cloud tier
# ---------------------------------------------------------------------------

print("\n## cloud tier")


@test("a cloud wrap writes the record with the provider update as destination")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n", "status": "IMPLEMENTING"})
    client = _FakeCloudClient()
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert len(client.calls) == 1, client.calls
    call = client.calls[0]
    assert call["projectId"] == "proj_1", "resolved exactly as the post used to"
    assert call["envelope"]["narrative"]["markdown"] == GOOD_BODY
    assert call["envelope"]["repo"] == "acme/widgets"

    update = _action(payload, "session_update")
    assert update["success"] is True and update["id"] == "rec_1", update
    assert update["permalinkPath"] == "/acme/activity/rec_1"
    assert not (session_dir / "focus.md").exists(), "a posted wrap clears state"


@test("the cloud tier keeps its own copy of every record")
def _():
    root, session_dir = _project()
    _wrap(root, _FakeCloudClient())
    record = _stream_lines(session_dir)[0]
    assert record["local"]["relay"] == {
        "state": "accepted", "id": "rec_1", "replayed": False,
    }, record["local"]


@test("a replayed record says so, locally and in the action")
def _():
    # The route answers 200 with the stored record when the same wrap arrives
    # twice. That is a success, and it is not a second session — the reader of
    # the stream should be able to tell.
    root, session_dir = _project()
    client = _FakeCloudClient(outcome={
        "success": True, "id": "rec_1", "permalinkPath": "/acme/activity/rec_1",
        "replayed": True, "providerUpdate": {"posted": False, "reason": "replayed"},
    })
    _code, payload, _err = _wrap(root, client)
    assert _action(payload, "session_update")["replayed"] is True
    assert _stream_lines(session_dir)[0]["local"]["relay"]["replayed"] is True
    assert client.fallback_calls == [], \
        "a replayed post is a fact about the destination, not a failure"


@test("a relay failure keeps the record locally and the session state intact")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    client = _FakeCloudClient(outcome=RelayError(0, "NETWORK_ERROR",
                                                 "unable to reach relay API", {}))
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert payload["success"] is False, payload
    record = _stream_lines(session_dir)[0]
    assert record["local"]["relay"]["state"] == "failed", record["local"]
    assert (session_dir / "focus.md").exists(), \
        "the handoff survives for the retry"


@test("a provider with no project updates still yields a wrapped session")
def _():
    # Jira and GitHub Issues have no project-update concept. Before the record
    # existed, that made "post the wrap" a Linear-only feature; now the record
    # is the write and the missing destination is a reported capability gap.
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    client = _FakeCloudClient(outcome={
        "success": True, "id": "rec_2", "permalinkPath": "/acme/activity/rec_2",
        "providerUpdate": {"posted": False, "reason": "unsupported"},
    })
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert payload["success"] is True, payload
    assert _action(payload, "project_update") is None
    assert not (session_dir / "focus.md").exists()


def _destination_failed(fallback=None):
    return _FakeCloudClient(
        outcome={
            "success": True, "id": "rec_3",
            "permalinkPath": "/acme/activity/rec_3",
            "providerUpdate": {"posted": False,
                               "error": "Linear rejected the update"},
        },
        fallback=fallback,
    )


@test("a destination that failed falls back to the project-update route")
def _():
    # The record is stored either way, so what is owed is the *post* — and
    # posting the narrative is what `/projects/update` still does. A wrap that
    # takes the fallback is a wrap that finished.
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    client = _destination_failed()
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert payload["success"] is True, payload
    assert len(client.fallback_calls) == 1, client.fallback_calls
    assert client.fallback_calls[0]["body"] == GOOD_BODY, "the same narrative"
    assert client.fallback_calls[0]["projectId"] == "proj_1", \
        "resolved the same way the record's destination was"
    posted = _action(payload, "project_update")
    assert posted["success"] is True and posted["viaFallback"] is True, posted
    assert posted["destinationError"] == "Linear rejected the update"
    assert not (session_dir / "focus.md").exists(), \
        "the update reached the team, so the wrap is done"


@test("a destination that cannot receive updates triggers no fallback")
def _():
    # `reason` is a fact about the destination — unsupported, or already
    # posted. Retrying it through another door would post twice, or post to a
    # provider that has nowhere to put it.
    for reason in ("unsupported", "replayed"):
        root, session_dir = _project({"focus.md": "FLY-1411\n"})
        client = _FakeCloudClient(outcome={
            "success": True, "id": "rec_4",
            "permalinkPath": "/acme/activity/rec_4",
            "providerUpdate": {"posted": False, "reason": reason},
        })
        code, payload, err = _wrap(root, client)
        assert code == 0, err
        assert payload["success"] is True, (reason, payload)
        assert client.fallback_calls == [], (reason, client.fallback_calls)
        assert _action(payload, "project_update") is None, reason
        assert not (session_dir / "focus.md").exists(), reason


@test("both routes shut leaves the state for a retry, record still appended")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    client = _destination_failed(fallback=RuntimeError("relay refused the post"))
    code, payload, err = _wrap(root, client)

    assert code == 0, err
    assert payload["success"] is False, payload
    failure = _action(payload, "project_update")
    assert failure["success"] is False and failure["viaFallback"] is True
    assert failure["error"] == "relay refused the post"
    assert failure["destinationError"] == "Linear rejected the update"
    assert (session_dir / "focus.md").exists(), \
        "nothing reached the team, so the session is not wrapped"
    lines = _stream_lines(session_dir)
    assert len(lines) == 1, "the record landed and is kept"
    assert lines[0]["local"]["relay"]["state"] == "accepted", lines[0]["local"]


# ---------------------------------------------------------------------------
# The transport (real client, no stubs)
# ---------------------------------------------------------------------------

print("\n## transport")


@test("an unreachable relay raises to the caller instead of exiting")
def _():
    # `fail()` raises SystemExit, which no `except Exception` catches — so
    # before this, the one caller with a recovery for an unreachable relay
    # (this wrap, whose whole job is not to lose the record) never got to run
    # it. Exercised against the real HTTP layer, not a fake that raises what
    # the test wants.
    root, _ = _project()
    client = _real_cloud_client(root)
    try:
        client.session_update_create({"schemaVersion": 1})
    except SystemExit as exit_error:  # pragma: no cover - the bug
        raise AssertionError(
            f"the client exited ({exit_error}) instead of raising"
        )
    except RelayError as err:
        assert err.code == "NETWORK_ERROR", err.code
        assert err.status == 0, err.status
    else:  # pragma: no cover - unreachable target answered
        raise AssertionError("the discard port answered")


@test("a wrap against an unreachable relay still keeps the record locally")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    code, payload, err = _wrap(root, _real_cloud_client(root))

    assert code == 0, f"the wrap exited: {err}"
    assert payload["success"] is False, payload
    lines = _stream_lines(session_dir)
    assert len(lines) == 1, "the record survives a network failure"
    assert lines[0]["local"]["relay"]["state"] == "failed", lines[0]["local"]
    assert (session_dir / "focus.md").exists(), "the handoff survives the retry"


@test("the operation id ignores the clock, so a respawn replays one record")
def _():
    # The id is derived from the request body under a seed. A session record
    # embeds `window.endedAt = now()`, so hashing the whole body gave a
    # respawned bridge a fresh id — a second server record and a second
    # provider post for one wrap.
    import flydocs_api
    root, _ = _project()
    client = _real_cloud_client(root)
    seen = []

    def fake_urlopen(request, timeout=None):
        headers = {k.lower(): v for k, v in request.headers.items()}
        seen.append(headers.get("x-operation-id"))
        return _Response({"success": True, "id": "rec"})

    def envelope(ended_at, session_id="2026-08-29-1", narrative="one"):
        return session.compose_session_update(
            repo="acme/widgets", session_id=session_id, actor={},
            started_at=ended_at, ended_at=ended_at, health="onTrack",
            issues=[], pending=[], blockers=[], narrative=narrative,
            visibility="team",
        )

    flydocs_api.set_operation_seed("op-seed-1")
    try:
        with patch("urllib.request.urlopen", fake_urlopen):
            client.session_update_create(envelope(1_000_000))
            # A retry composes a later clock, and — if the first attempt got as
            # far as appending — the next sequence number too. Neither is the
            # intent.
            client.session_update_create(
                envelope(9_999_999, session_id="2026-08-29-2"))
            client.session_update_create(envelope(9_999_999, narrative="two"))
    finally:
        flydocs_api.set_operation_seed(None)

    assert seen[0] == seen[1], \
        f"two attempts at one wrap must key alike: {seen[0]} vs {seen[1]}"
    assert seen[2] != seen[0], \
        "a different wrap is a different intent and must key differently"


# ---------------------------------------------------------------------------
# The wire body (route contract, FLY-1410 §2)
# ---------------------------------------------------------------------------

print("\n## wire body")


@test("provenance carries the source and version only — never an operationId")
def _():
    # The route reads `X-Operation-Id` from the header and answers a body that
    # also claims one with a 400 finding. The id is transport, not provenance.
    root, session_dir = _project()
    client = _FakeCloudClient()
    _wrap(root, client)
    sent = client.calls[0]["envelope"]["provenance"]
    assert set(sent) <= {"source", "cliVersion"}, sent
    assert sent["source"] == "cli"
    assert "operationId" not in sent
    assert "recordedAt" not in sent, "the server stamps that one"
    assert "operationId" not in _stream_lines(session_dir)[0]["provenance"]


@test("the body's repo is the value the X-Repo header carries")
def _():
    # The route refuses a body that names a different repo than the header. The
    # header comes from the relay's own slug, so that is what the envelope has
    # to say — not the config value it might disagree with.
    import flydocs_api
    root, session_dir = _project()
    client = _real_cloud_client(root)
    client._relay.repo_slug = "acme/widgets-as-git-knows-it"
    seen = []

    def fake_urlopen(request, timeout=None):
        headers = {k.lower(): v for k, v in request.headers.items()}
        seen.append((headers.get("x-repo"),
                     json.loads(request.data.decode())["repo"]))
        return _Response({"success": True, "id": "rec",
                          "permalinkPath": "/a/activity/rec"})

    with patch("urllib.request.urlopen", fake_urlopen):
        code, _payload, err = _wrap(root, client)

    assert code == 0, err
    header_repo, body_repo = seen[0]
    assert header_repo == "acme/widgets-as-git-knows-it", header_repo
    assert body_repo.lower() == header_repo.lower(), (body_repo, header_repo)
    assert body_repo != "acme/widgets", \
        "config's slug lost to the one the request actually claims"
    assert _stream_lines(session_dir)[0]["repo"] == body_repo


@test("blank pending and blocker entries never reach the wire")
def _():
    # A shell loop that emits `--pending ""` used to 400 the whole wrap: the
    # route refuses a blank entry, and it is right to.
    root, session_dir = _project()
    client = _FakeCloudClient()
    code, payload, err = _wrap(
        root, client,
        pending=["ship it", "", "   ", "\n"],
        blockers=["", "waiting on Kyle", " "],
    )

    assert code == 0, err
    envelope = client.calls[0]["envelope"]
    assert envelope["pending"] == ["ship it"], envelope["pending"]
    assert envelope["blockers"] == [{"text": "waiting on Kyle"}], \
        envelope["blockers"]
    assert payload["summary"]["pendingCount"] == 1
    assert payload["summary"]["blockerCount"] == 1
    assert _stream_lines(session_dir)[0]["pending"] == ["ship it"]


# ---------------------------------------------------------------------------
# Deterministic rejections
# ---------------------------------------------------------------------------

print("\n## deterministic rejections")


@test("a 4xx envelope rejection appends nothing and keeps the state")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    error = RelayError(400, "VALIDATION_ERROR", "repo is not in this workspace",
                       {"code": "VALIDATION_ERROR",
                        "error": "repo is not in this workspace",
                        "findings": [{"field": "repo", "message": "unknown"}]})
    code, _payload, err = _wrap(root, _FakeCloudClient(outcome=error))

    assert code == 1, "a decision the server will repeat is not a retry"
    assert _stream_lines(session_dir) == [], \
        "appending it would leave one record per retry, all refused"
    assert (session_dir / "focus.md").exists()
    assert "VALIDATION_ERROR" in err and "repo (unknown)" in err, err


@test("an expired key is refused the same way, not recorded as transient")
def _():
    root, session_dir = _project()
    error = RelayError(401, "UNAUTHORIZED", "API key is not valid", {})
    code, _payload, err = _wrap(root, _FakeCloudClient(outcome=error))
    assert code == 1, err
    assert _stream_lines(session_dir) == []


@test("a throttle or an in-flight operation stays transient")
def _():
    # 429 and 409 are "later", not "no" — the record is kept locally and the
    # session state survives for the retry.
    for status, code_name in ((429, "RATE_LIMITED"), (409, "OPERATION_IN_FLIGHT")):
        root, session_dir = _project({"focus.md": "FLY-1411\n"})
        error = RelayError(status, code_name, "try later", {})
        code, payload, err = _wrap(root, _FakeCloudClient(outcome=error))
        assert code == 0, f"{code_name}: {err}"
        assert payload["success"] is False
        assert len(_stream_lines(session_dir)) == 1, code_name
        assert (session_dir / "focus.md").exists(), code_name


@test("a local post that failed appends nothing")
def _():
    # The local tier has no server-side identity to collapse two copies, so an
    # append after a failed write is a duplicate the next wrap cannot detect.
    class _Broken(_FakeLocalClient):
        def project_update(self, *_a, **_k):
            raise OSError("read-only file system")

    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    code, payload, err = _wrap(root, _Broken())
    assert code == 0, err
    assert payload["success"] is False
    assert _stream_lines(session_dir) == []
    assert (session_dir / "focus.md").exists()


# ---------------------------------------------------------------------------
# The stream file itself
# ---------------------------------------------------------------------------

print("\n## stream.jsonl")


@test("the stream is append-only across wraps and survives cleanup")
def _():
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    _wrap(root, _FakeLocalClient())
    (session_dir / "focus.md").write_text("FLY-9\n")
    _wrap(root, _FakeLocalClient(), issues=["FLY-9"])

    lines = _stream_lines(session_dir)
    assert len(lines) == 2, lines
    assert [entry["issues"][0]["ref"] for entry in lines] == ["FLY-1411", "FLY-9"]
    assert not (session_dir / "focus.md").exists(), "cleanup still ran"


@test("no cleanup path may name a stream file")
def _():
    for name in ("stream.jsonl", "stream.1.jsonl", "stream.3.jsonl"):
        assert session.is_stream_file(name), name
    for name in ("focus.md", "status", "status-ref", "acceptance-criteria.md",
                 "last-summary.json", "streams.jsonl", "stream.jsonl.bak"):
        assert not session.is_stream_file(name), name


@test("the stream rotates at the size cap and keeps three generations")
def _():
    root, session_dir = _project()
    path = session_dir / "stream.jsonl"
    path.write_text("x" * session.STREAM_MAX_BYTES)
    (session_dir / "stream.1.jsonl").write_text("gen1\n")
    (session_dir / "stream.2.jsonl").write_text("gen2\n")
    (session_dir / "stream.3.jsonl").write_text("gen3\n")

    _wrap(root, _FakeLocalClient())

    assert len(_stream_lines(session_dir)) == 1, "the live file starts fresh"
    assert (session_dir / "stream.1.jsonl").stat().st_size >= \
        session.STREAM_MAX_BYTES, "the full file became generation 1"
    assert (session_dir / "stream.2.jsonl").read_text() == "gen1\n"
    assert (session_dir / "stream.3.jsonl").read_text() == "gen2\n"
    assert not (session_dir / "stream.4.jsonl").exists(), \
        "three generations, and the oldest is dropped"


@test("the append is one os.write to an O_APPEND descriptor")
def _():
    # A record is far larger than the 8 KiB a text-mode buffer flushes in, so
    # `open('a').write()` can interleave two concurrent wraps mid-line. One
    # write to an append-only descriptor cannot.
    root, session_dir = _project()
    flags, writes = [], []
    real_open, real_write = os.open, os.write

    def spy_open(path, flag, *args):
        if str(path).endswith("stream.jsonl"):
            flags.append(flag)
        return real_open(path, flag, *args)

    def spy_write(fd, data):
        writes.append(data)
        return real_write(fd, data)

    big = GOOD_BODY + "\n" + ("- a long line of standup prose. " * 400)
    with patch.object(os, "open", spy_open), patch.object(os, "write", spy_write):
        code, _payload, err = _wrap(root, _FakeLocalClient(), body=big)

    assert code == 0, err
    assert flags, "the stream was not opened through os.open"
    flag = flags[0]
    assert flag & os.O_APPEND and flag & os.O_CREAT and flag & os.O_WRONLY, flag
    record_writes = [w for w in writes if b'"schemaVersion"' in w]
    assert len(record_writes) == 1, "the line reached the file in one write"
    assert len(record_writes[0]) > 8192, "and it is past a text buffer's flush"
    assert record_writes[0].endswith(b"\n")
    assert len(_stream_lines(session_dir)) == 1


@test("losing the rotation race is not a failed wrap")
def _():
    # Rotation runs after the relay has accepted. An exception here would
    # report a wrap that did happen as failed, and invite a retry that
    # duplicates the record.
    root, session_dir = _project()
    (session_dir / "stream.jsonl").write_text("x" * session.STREAM_MAX_BYTES)
    real_replace = Path.replace

    def losing_replace(self, target):
        if self.name == "stream.jsonl":
            raise FileNotFoundError(2, "No such file or directory", str(self))
        return real_replace(self, target)

    with patch.object(Path, "replace", losing_replace):
        code, payload, err = _wrap(root, _FakeCloudClient())

    assert code == 0, err
    assert payload["success"] is True, payload
    assert _action(payload, "session_stream")["success"] is True
    assert (session_dir / "stream.jsonl").exists()


@test("a file under the cap is never rotated")
def _():
    root, session_dir = _project()
    _wrap(root, _FakeLocalClient())
    _wrap(root, _FakeLocalClient())
    assert not (session_dir / "stream.1.jsonl").exists()
    assert len(_stream_lines(session_dir)) == 2


# ---------------------------------------------------------------------------
# Graph reconciliation
# ---------------------------------------------------------------------------

print("\n## graph reconciliation")


@test("the graph node and the record name the same session")
def _():
    root, session_dir = _project()
    code, payload, err = _wrap(root, _FakeLocalClient(), notes="Shipped the record.")
    assert code == 0, err
    graph = _action(payload, "graph_record")
    assert graph is not None and graph["success"] is True, payload["actions"]
    assert graph["matchesRecord"] is True, graph
    assert "graphSessionId" not in graph, \
        "the id is only named when it disagrees — it changes every day"
    nodes = json.loads(
        (root / "flydocs" / "context" / "graph.json").read_text())["nodes"]
    assert f"session:{_stream_lines(session_dir)[0]['sessionId']}" in nodes, nodes


@test("a wrap that did not land writes no graph node; the retry writes one")
def _():
    # A node per *attempt* is a graph that counts failures as sessions. The
    # node is gated on the same condition as cleanup: did this wrap land?
    root, session_dir = _project({"focus.md": "FLY-1411\n"})
    graph_file = root / "flydocs" / "context" / "graph.json"

    failed = _FakeCloudClient(
        outcome=RelayError(0, "NETWORK_ERROR", "unable to reach relay API", {}))
    code, payload, err = _wrap(root, failed)
    assert code == 0, err
    assert payload["success"] is False
    assert _action(payload, "graph_record") is None, payload["actions"]
    assert not graph_file.exists(), "no node for a wrap that did not land"

    code, payload, err = _wrap(root, _FakeCloudClient())
    assert code == 0, err
    assert _action(payload, "graph_record")["success"] is True
    nodes = json.loads(graph_file.read_text())["nodes"]
    sessions = [key for key in nodes if key.startswith("session:")]
    assert len(sessions) == 1, sessions


@test("graph_session refuses a session id that is not one")
def _():
    import subprocess
    result = subprocess.run(
        [sys.executable, str(SCRIPT_DIR / "graph_session.py"),
         "--summary", "x", "--session-id", "../../etc/passwd",
         "--root", str(_project()[0])],
        capture_output=True, text=True, timeout=30,
    )
    assert result.returncode == 1, result.stdout
    assert "YYYY-MM-DD" in result.stderr, result.stderr


@test("two wraps with no notes get distinct ids, and the graph gets both")
def _():
    # The primary path: `session_wrap` over MCP sends no notes. The graph node
    # used to be skipped entirely for those, so nothing advanced the sequence
    # and both wraps of a day claimed the same session id — one record on the
    # server, the second superseding the first.
    root, session_dir = _project()
    first = _wrap(root, _FakeLocalClient())
    second = _wrap(root, _FakeLocalClient())
    assert first[0] == 0 and second[0] == 0, (first[2], second[2])

    ids = [line["sessionId"] for line in _stream_lines(session_dir)]
    today = _dt.date.today().isoformat()
    assert ids == [f"{today}-1", f"{today}-2"], ids

    graph = json.loads((root / "flydocs" / "context" / "graph.json").read_text())
    for index, payload in enumerate((first[1], second[1])):
        action = _action(payload, "graph_record")
        assert action is not None, payload["actions"]
        assert action["matchesRecord"] is True, action
        assert f"session:{ids[index]}" in graph["nodes"], graph["nodes"].keys()
    labels = {
        node["label"] for key, node in graph["nodes"].items()
        if key.startswith("session:")
    }
    assert labels == {"Accomplished — The record landed."}, \
        f"the wrap body labels the node when there are no notes: {labels}"


# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------

print(f"\n{'='*50}")
print(f"Results: {passed} passed, {failed} failed, {passed + failed} total")
if failed > 0:
    sys.exit(1)