atlas-iac/testing/tests/test_hermes_hux_contract_events.py

"""HUX-01 activity events: ordering, idempotency, replay, reconnect, redaction, cancellation.

Security obligations exercised: SO-10..SO-19 (server-set provenance, detail
allowlist, secret scrub, size caps, serve-time redaction, server-assigned seq,
idempotent replay after ownership, bounded pages and streams, 404 for foreign
conversations, evidence URI allowlist) and SO-28 (private mode writes nothing).
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import audit, contracts, errors, events, identity, redaction, store  # noqa: E402
from hux.server import build_router  # noqa: E402

SCHEMAS = contracts.load_all()
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
CONV = "conv_0001abcd"


def ident(**overrides) -> identity.Identity:
    base = {"tenant_slot": "slot-3", "subject": "usr_0123456789abcdef", "surface": "chat", "trust": "router"}
    return identity.Identity(**{**base, **overrides})


def router_for(tmp_path):
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})


def call(router, method, path, headers=HEADERS, body=None):
    raw = json.dumps(body).encode() if body is not None else b""
    response = router.dispatch(method, path, headers, raw)
    return response.status, response.body, response


def tenant(tmp_path, who=None) -> store.TenantStore:
    return store.TenantStore(tmp_path, who or ident())


def valid(record) -> None:
    assert contracts.validate_record(record, SCHEMAS) == [], record


# --- emit: ordering and idempotency (SO-15, SO-16) ---------------------------------

def test_emit_assigns_monotonic_seq_and_validates(tmp_path):
    s = tenant(tmp_path)
    first = events.emit(s, ident(), CONV, "message.user", "hello", {"message_id": "m1", "chars": 5, "junk": "x"}, turn=1)
    second = events.emit(s, ident(), CONV, "message.assistant", "hi", run_id="run_9f", correlation_id="c1")
    assert (first["seq"], second["seq"]) == (1, 2)
    assert first["id"] != second["id"] and first["id"].startswith("evt_")
    assert first["detail"] == {"message_id": "m1", "chars": 5}
    assert second["run_id"] == "run_9f" and second["provenance"]["run_id"] == "run_9f"
    assert first["provenance"]["actor"] == {"type": "user", "id": "usr_0123456789abcdef"}
    assert second["provenance"]["actor"]["type"] == "assistant"
    assert first["identity"] == ident().record()
    for record in (first, second):
        valid(record)
    checkpoint = s.get(events.SEQ_FAMILY, f"seq_{CONV}")
    assert checkpoint["next_seq"] == 3 and checkpoint["last_event_id"] == second["id"]
    other = events.emit(s, ident(), "conv_0002abcd", "run.started", "another conversation")
    assert other["seq"] == 1


def test_emit_is_idempotent_on_key(tmp_path):
    s = tenant(tmp_path)
    one = events.emit(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1")
    again = events.emit(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1")
    assert again == one
    assert len(s.read(events.FAMILY, CONV)) == 1
    _, replayed = events.emit_with_status(s, ident(), CONV, "run.started", "start", idempotency_key="run_9f:start:1")
    assert replayed is True


def test_emit_rejects_unknown_kind_and_sensitivity(tmp_path):
    s = tenant(tmp_path)
    with pytest.raises(errors.Invalid):
        events.emit(s, ident(), CONV, "not.a.kind", "x")
    with pytest.raises(errors.Invalid):
        events.emit(s, ident(), CONV, "run.started", "x", sensitivity="secret")
    with pytest.raises(errors.Invalid):
        events.emit(s, ident(), "not a conversation id", "run.started", "x")


def test_emit_validates_against_contract_before_writing(tmp_path, monkeypatch):
    s = tenant(tmp_path)
    monkeypatch.setattr(events, "new_id", lambda prefix: "bad id")
    with pytest.raises(errors.Invalid):
        events.emit(s, ident(), CONV, "run.started", "x")
    assert s.read(events.FAMILY, CONV) == []


def test_worker_emits_as_system_actor(tmp_path):
    s = tenant(tmp_path)
    record = events.emit(s, ident(surface="worker", trust="worker"), CONV, "message.user", "x")
    assert record["provenance"]["actor"] == {"type": "system", "id": "hux-worker"}


def test_private_mode_writes_nothing(tmp_path):
    s = tenant(tmp_path)
    s.put("conversations", {"id": "conv_priv0001", "mode": "private"})
    assert events.emit(s, ident(), "conv_priv0001", "message.user", "secret chat") is None
    assert s.read(events.FAMILY, "conv_priv0001") == []
    router = router_for(tmp_path)
    status, body, _ = call(router, "POST", "/hux/v1/conversations/conv_priv0001/events", body={"kind": "message.user", "summary": "x"})
    assert status == 204 and body is None


# --- redaction pipeline (SO-11, SO-12, SO-13, SO-19) ------------------------------

def test_tool_call_detail_keeps_only_hash_and_names(tmp_path):
    s = tenant(tmp_path)
    detail = {
        "tool": "shell", "capability": "shell", "arguments": {"cmd": "curl -H 'Authorization: Bearer abcdefghijklmnopqrstuv'"},
        "argument_names": ["cmd", "cwd", "cmd"], "argument_hash": "a" * 64, "argument_bytes": 61,
        "target_path": "/opt/data/.env", "stdout": "leak",
    }
    record = events.emit(s, ident(), CONV, "tool.call", "ran shell", detail)
    assert record["detail"] == {"tool": "shell", "capability": "shell", "argument_names": ["cmd", "cwd"], "argument_hash": "a" * 64, "argument_bytes": 61}
    assert "arguments" not in json.dumps(record) and "leak" not in json.dumps(record)
    kept = events.emit(s, ident(), CONV, "tool.call", "wrote", {"tool": "write", "target_path": "/opt/data/workspace/notes.md"})
    assert kept["detail"]["target_path"] == "/opt/data/workspace/notes.md"
    plan = events.emit(s, ident(), CONV, "decision.plan", "plan", {"steps": ["s" * 300] * 25})
    assert len(plan["detail"]["steps"]) == 20 and len(plan["detail"]["steps"][0]) == 200
    assert events.emit(s, ident(), CONV, "decision.plan", "plan", {"steps": "nope"})["detail"] == {"steps": []}
    assert events.emit(s, ident(), CONV, "tool.call", "x", {"tool": "t", "argument_names": "nope"})["detail"]["argument_names"] == []


@pytest.mark.parametrize("secret,klass", [
    ("Authorization: Bearer abcdefghijklmnopqrstuvwxyz0123", "bearer"),
    ("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0In0.abcdef", "jwt"),
    ("sk-abcdefghijklmnopqrstuvwxyz1234", "openai_key"),
    ("ghp_abcdefghijklmnopqrstuvwxyz1234", "github_token"),
    ("glpat-abcdefghijklmnopqrstuv", "gitlab_token"),
    ("AKIAABCDEFGHIJKLMNOP", "aws_key"),
    ("xoxb-1234567890-abcdef", "slack_token"),
    ("hvs.CAESIJabcdefghijklmnopqrstuvwxyz", "vault_token"),
    ("password=hunter2hunter2", "password"),
    ("0123456789abcdef0123456789abcdef", "hex"),
    ("-----BEGIN RSA PRIVATE KEY-----\nMIIE\n-----END RSA PRIVATE KEY-----", "private_key"),
])
def test_secret_patterns_are_scrubbed(tmp_path, secret, klass):
    s = tenant(tmp_path)
    record = events.emit(s, ident(), CONV, "decision.route", f"used {secret} today", {"reason": secret})
    text = json.dumps(record)
    assert secret not in text and f"[redacted:{klass}]" in text
    assert record["redaction"]["level"] == "partial" and klass in record["redaction"]["reason"]


def test_canaries_from_environment_and_file_are_scrubbed(tmp_path, monkeypatch):
    canary_file = tmp_path / "env"
    canary_file.write_text("OPENAI_KEY='filecanaryvalue'\nSHORT=x\n")
    monkeypatch.setenv("HUX_RELAY_KEY", "relaykeycanary")
    monkeypatch.setenv("HUX_CANARY_FILE", str(canary_file))
    text, hits = redaction.scrub_text("relay relaykeycanary file filecanaryvalue")
    assert text == "relay [redacted:canary] file [redacted:canary]" and hits == ["canary", "canary"]
    assert redaction.canaries({}) == []


def test_hashes_survive_the_hex_rule(tmp_path):
    s = tenant(tmp_path)
    record = events.emit(s, ident(), CONV, "artifact.version", "v2", {"artifact_id": "art_0001aaaa", "hash": "sha256:" + "b" * 64}, [{"kind": "artifact_version", "id": "art_0001aaaa@2", "hash": "sha256:" + "b" * 64}])
    assert record["detail"]["hash"] == "sha256:" + "b" * 64 and record["evidence"][0]["hash"] == "sha256:" + "b" * 64
    assert record["redaction"] == {"level": "none"}


def test_oversized_detail_is_truncated_and_line_cap_enforced(tmp_path):
    s = tenant(tmp_path)
    record = events.emit(s, ident(), CONV, "decision.route", "big", {"reason": "r" * 40000})
    assert record["detail"]["truncated"] is True and record["detail"]["bytes"] > redaction.DETAIL_CAP_BYTES
    assert record["redaction"]["level"] == "partial" and "truncated" in record["redaction"]["reason"]
    huge = [{"kind": "url", "id": "u", "uri": "https://x/" + "z" * 1990} for _ in range(64)]
    with pytest.raises(errors.TooLarge):
        events.emit(s, ident(), CONV, "decision.route", "too big", evidence=huge)


def test_evidence_uris_outside_allowlist_are_dropped(tmp_path):
    s = tenant(tmp_path)
    evidence = [
        {"kind": "file", "id": "env", "uri": "file:///opt/data/.env"},
        {"kind": "url", "id": "doc", "uri": "https://example.test/doc"},
        {"kind": "artifact_version", "id": "art_0001aaaa@1", "uri": "artifact://art_0001aaaa/1"},
        "garbage", {"kind": 5, "id": "x"},
    ]
    record = events.emit(s, ident(), CONV, "tool.result", "read", evidence=evidence)
    assert [ref.get("uri") for ref in record["evidence"]] == [None, "https://example.test/doc", "artifact://art_0001aaaa/1"]
    assert len(redaction.filter_evidence([{"kind": "run", "id": "r"}] * 100)) == 64


def test_redaction_level_derivation():
    assert redaction.derive_level("personal", [], False) == {"level": "none"}
    assert redaction.derive_level("sensitive", [], False)["level"] == "partial"
    assert redaction.derive_level("sensitive", ["hex"], True)["level"] == "partial"
    assert redaction.derive_level("restricted", ["hex"], True) == {"level": "full", "reason": "restricted content"}
    assert redaction.derive_level("public", [], True)["reason"] == "detail truncated"


def test_serve_time_redaction_by_surface(tmp_path):
    s = tenant(tmp_path)
    record = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"})
    assert redaction.redact_record({**record, "_meta": {"x": 1}}, "chat") == record
    for surface in ("telegram", "voice"):
        served = redaction.redact_record(record, surface)
        assert "detail" not in served and served["redaction"]["level"] == "partial"
        valid(served)
    partial = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"}, sensitivity="sensitive")
    assert "detail" not in redaction.redact_record(partial, "chat")
    full = events.emit(s, ident(), CONV, "decision.route", "routed", {"requested": "fast"}, ["x"], sensitivity="restricted", run_id="r")
    served = redaction.redact_record(full, "chat")
    assert served["summary"] == "[redacted]" and "detail" not in served and "evidence" not in served and "run_id" not in served
    valid(served)
    assert redaction.redact_record({"schema": "x", "content": "c"}, "chat") == {"schema": "x", "content": "c"}
    memory_like = {"schema": "hux.memory.v1", "content": "secret", "redaction": {"level": "full"}}
    assert redaction.redact_record(memory_like, "chat")["content"] == ""


def test_full_rewrite_keeps_timeline_contiguous(tmp_path):
    s = tenant(tmp_path)
    events.emit(s, ident(), CONV, "message.user", "one", {"message_id": "m1"}, idempotency_key="m1:00000001")
    events.emit(s, ident(), CONV, "message.user", "two", sensitivity="sensitive")
    assert events.rewrite_full(s, CONV, "", "forgotten") == 2
    rows = s.read(events.FAMILY, CONV)
    assert [r["seq"] for r in rows] == [1, 2] and all(r["redaction"]["level"] == "full" for r in rows)
    assert rows[0]["summary"] == "[redacted]" and rows[0]["idempotency_key"] == "m1:00000001" and "detail" not in rows[0]
    for row in rows:
        valid(row)
    assert events.rewrite_full(s, CONV, "", "again") == 0


def test_memory_reference_redaction_spans_conversations(tmp_path):
    s = tenant(tmp_path)
    events.emit(s, ident(), CONV, "memory.committed", "kept", {"memory_id": "mem_0001aaaa"})
    events.emit(s, ident(), "conv_0002abcd", "message.user", "plain", evidence=[{"kind": "memory", "id": "mem_0001aaaa"}])
    events.emit(s, ident(), "conv_0002abcd", "message.user", "unrelated")
    assert events.redact_memory_references(s, "mem_0001aaaa") == 2
    assert s.read(events.FAMILY, "conv_0002abcd")[1]["redaction"]["level"] == "none"


# --- routes -------------------------------------------------------------------------

def test_list_pages_with_after_seq_and_limit(tmp_path):
    router = router_for(tmp_path)
    s = tenant(tmp_path)
    for n in range(5):
        events.emit(s, ident(), CONV, "message.user", f"m{n}")
    status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=2")
    assert status == 200 and [e["seq"] for e in body["items"]] == [1, 2] and body["next"] == 2
    status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=2&limit=2")
    assert [e["seq"] for e in body["items"]] == [3, 4] and body["next"] == 4
    status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=4&limit=2")
    assert [e["seq"] for e in body["items"]] == [5] and body["next"] is None
    for item in body["items"]:
        valid(item)
    status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=5000")
    assert len(body["items"]) == 5
    assert events.PAGE_MAX == 200
    assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=abc")[0] == 400
    assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events?limit=0&after_seq=-5")[1]["items"][0]["seq"] == 1


def test_foreign_conversation_is_404_not_403(tmp_path):
    router = router_for(tmp_path)
    events.emit(tenant(tmp_path), ident(), CONV, "message.user", "mine")
    status, body, _ = call(router, "GET", f"/hux/v1/conversations/{CONV}/events", headers=OTHER)
    assert (status, body["code"]) == (404, "not_found")
    assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0", headers=OTHER)[0] == 404
    assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers=OTHER, body={"kind": "run.started", "summary": "x"})[0] == 404
    assert call(router, "GET", "/hux/v1/conversations/conv_nope0000/events")[0] == 404
    rows = audit.recent(tenant(tmp_path, ident(subject="usr_fedcba9876543210")))
    assert {r["outcome"] for r in rows} == {"not_found"}


def test_post_event_server_assigns_and_replays(tmp_path):
    router = router_for(tmp_path)
    tenant(tmp_path).put("conversations", {"id": CONV, "mode": "fast"})
    body = {"kind": "run.cancelled", "summary": "stopped by user", "detail": {"run_id": "run_9f", "outcome": "cancelled", "receipt_id": "rcpt_0001aaaa", "raw": 1},
            "evidence": [{"kind": "approval", "id": "apr_0001aaaa"}, {"kind": "run", "id": "run_9f"}], "run_id": "run_9f", "turn": 3,
            "identity": {"tenant_slot": "slot-9"}, "provenance": {"surface": "worker"}}
    headers = {**HEADERS, "Idempotency-Key": "run_9f:cancel:1"}
    status, first, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers, body)
    assert status == 201 and first["seq"] == 1 and first["identity"]["tenant_slot"] == "slot-3" and first["provenance"]["surface"] == "chat"
    assert first["provenance"]["actor"]["type"] == "user" and first["detail"] == {"run_id": "run_9f", "outcome": "cancelled", "receipt_id": "rcpt_0001aaaa"}
    assert [e["kind"] for e in first["evidence"]] == ["approval", "run"]
    valid(first)
    status, again, response = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", headers, body)
    assert status == 200 and again == first and response.headers["HUX-Replayed"] == "true"
    assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "seq": 9})[0] == 400
    assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "id": "evt_0000"})[0] == 400
    assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"summary": "x"})[0] == 400
    assert call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, [1])[0] == 400
    status, body, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/events", HEADERS, {"kind": "run.started", "summary": "x", "detail": "nope", "evidence": "nope", "turn": "3"})
    assert status == 201 and body["turn"] == 0 and "detail" not in body
    reasons = [r["reason"] for r in audit.recent(tenant(tmp_path)) if r["action"] == "events.append" and r["outcome"] == "allow" and r.get("reason")]
    assert reasons == ["replayed"]


def _sse_records(chunks):
    out = []
    for chunk in chunks:
        for line in chunk.decode().splitlines():
            if line.startswith("data: "):
                out.append(json.loads(line[6:]))
    return out


def test_stream_replays_then_polls_and_resumes_from_last_event_id(tmp_path):
    router = router_for(tmp_path)
    s = tenant(tmp_path)
    for n in range(3):
        events.emit(s, ident(), CONV, "message.user", f"m{n}")
    status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0")
    chunks = list(response.stream())
    assert status == 200 and chunks[0] == b"retry: 2000\n\n"
    assert [r["seq"] for r in _sse_records(chunks)] == [1, 2, 3]
    assert b"id: 3\nevent: message.user\n" in chunks[3]
    status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=1&poll_ms=0", {**HEADERS, "Last-Event-ID": "2"})
    generator = response.stream()
    assert next(generator) == b"retry: 2000\n\n"
    assert _sse_records([next(generator)])[0]["seq"] == 3
    assert next(generator) == b": keepalive\n\n"
    events.emit(s, ident(), CONV, "message.assistant", "late")
    assert _sse_records([next(generator)])[0]["seq"] == 4
    assert list(generator) == []
    status, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?after_seq=4&max_polls=0", {**HEADERS, "X-Hux-Surface": "voice"})
    assert _sse_records(list(response.stream())) == []


def test_second_stream_closes_the_first(tmp_path):
    router = router_for(tmp_path)
    events.emit(tenant(tmp_path), ident(), CONV, "message.user", "m")
    _, _, first = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=5&poll_ms=0")
    first_gen = first.stream()
    next(first_gen)
    next(first_gen)
    _, _, second = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0")
    list(second.stream())
    assert list(first_gen) == []


def test_stream_redacts_for_voice_surface(tmp_path):
    router = router_for(tmp_path)
    events.emit(tenant(tmp_path), ident(), CONV, "decision.route", "r", {"requested": "fast"})
    _, _, response = call(router, "GET", f"/hux/v1/conversations/{CONV}/events/stream?max_polls=0", {**HEADERS, "X-Hux-Surface": "voice"})
    records = _sse_records(list(response.stream()))
    assert "detail" not in records[0] and records[0]["redaction"]["level"] == "partial"


def test_flag_off_hides_event_routes(tmp_path):
    router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation", "HUX_ROUTER_KEY": "rk"})
    assert call(router, "GET", f"/hux/v1/conversations/{CONV}/events")[1]["code"] == "flag_off"


def test_f5_crash_between_append_and_checkpoint_never_duplicates_a_seq(tmp_path, monkeypatch):
    """F5 / SO-15: with the checkpoint left behind the ledger, the next emit takes max(checkpoint, tail + 1) so paging stays complete."""
    router = router_for(tmp_path)
    s = tenant(tmp_path)
    events.emit(s, ident(), CONV, "message.user", "one")
    original = store.TenantStore.put

    def crash(self, family, record, expected_revision=None):
        if family == events.SEQ_FAMILY:
            raise RuntimeError("simulated crash after the ledger append")
        return original(self, family, record, expected_revision)

    monkeypatch.setattr(store.TenantStore, "put", crash)
    with pytest.raises(RuntimeError):
        events.emit(s, ident(), CONV, "message.user", "two")
    monkeypatch.undo()
    assert s.get(events.SEQ_FAMILY, events._seq_id(CONV))["next_seq"] == 2 and events.last_seq(s, CONV) == 2
    third = events.emit(s, ident(), CONV, "message.user", "three")
    assert third["seq"] == 3 and s.get(events.SEQ_FAMILY, events._seq_id(CONV))["next_seq"] == 4
    items = call(router, "GET", f"/hux/v1/conversations/{CONV}/events")[1]["items"]
    assert [(e["seq"], e["summary"]) for e in items] == [(1, "one"), (2, "two"), (3, "three")]
    assert [e["seq"] for e in call(router, "GET", f"/hux/v1/conversations/{CONV}/events?after_seq=2")[1]["items"]] == [3]


def test_f5_last_seq_reads_only_the_tail_and_skips_torn_lines(tmp_path):
    """F5: a torn trailing line (crash mid-append) and a missing ledger are both handled."""
    s = tenant(tmp_path)
    assert events.last_seq(s, "conv_none0001") == 0
    for index in range(3):
        events.emit(s, ident(), CONV, "message.user", f"m{index}")
    path = s.root / events.FAMILY / f"{CONV}.jsonl"
    with open(path, "ab") as handle:
        handle.write(b'{"schema":"hux.event.v1","seq":9')
    assert events.last_seq(s, CONV) == 3
    with open(path, "ab") as handle:
        handle.write(b"\n[]\n\n")
    assert events.last_seq(s, CONV) == 3
    path.write_bytes(b"\n\n")
    assert events.last_seq(s, CONV) == 0


def test_conversation_known_accepts_a_ledger_without_a_checkpoint(tmp_path):
    """F5 / SO-18: a conversation whose checkpoint write was lost is still this subject's conversation."""
    s = tenant(tmp_path)
    events.emit(s, ident(), CONV, "message.user", "one")
    s.delete(events.SEQ_FAMILY, events._seq_id(CONV))
    assert events.conversation_known(s, CONV) is True
    assert events.conversation_known(s, "conv_other0001") is False
    assert events.emit(s, ident(), CONV, "message.user", "two")["seq"] == 2


def test_f13d_target_path_is_normalised_before_the_workspace_check(tmp_path):
    """F13d / SO-11: ``workspace/../.env`` is not a workspace path; a dotted path inside the workspace is stored normalised."""
    s = tenant(tmp_path)
    escaped = events.emit(s, ident(), CONV, "tool.call", "ran", {"tool": "bash", "target_path": "/opt/data/workspace/../.env"})
    assert "target_path" not in escaped["detail"]
    inside = events.emit(s, ident(), CONV, "tool.call", "ran", {"tool": "bash", "target_path": "/opt/data/workspace/a/../notes.md"})
    assert inside["detail"]["target_path"] == "/opt/data/workspace/notes.md"
    assert "target_path" not in events.emit(s, ident(), CONV, "tool.call", "ran", {"tool": "bash", "target_path": 7})["detail"]
    assert "target_path" not in events.emit(s, ident(), CONV, "tool.call", "ran", {"tool": "bash", "target_path": "/opt/data/workspace"})["detail"]


def test_events_sources_stay_under_500_lines():
    for name in ("events.py", "redaction.py"):
        assert len((FOUNDATION / "hux" / name).read_text().splitlines()) <= 500, name