feat(collector): publish attacker.session.ended on session_recorded events

The TTP worker subscribes to attacker.session.ended but no upstream
component published it — the rule pack (R0001–R0030) therefore never
fired on live SSH traffic even after the consume-side wiring landed
in E.3.18a/b/c.

The collector now hosts a per-attacker_ip command index
(_SessionAggregator) that watches the same parsed-event stream as
_publish_log. Shell `command` events are appended to a per-IP list;
on `session_recorded` the aggregator slices the list to commands
inside the [ended_at - duration_s, ended_at] window and publishes
attacker.session.ended with the session metadata + commands list.
The TTP worker's _build_events fan-out (E.3.18b) turns each command
into a source_kind="command" TaggerEvent that the RuleEngineTagger
(E.3.18c) matches against R0001–R0030.

Memory bound: per-IP entries TTL-evict at DECNET_COLLECTOR_SESSION_AGG_TTL_SEC
(default 3600 s). Publish failures are swallowed in the aggregator —
a misbehaving bus cannot stall the per-container stream threads.

tests/collector/test_session_ended_publish.py

@@ -0,0 +1,82 @@
+"""Collector publishes ``attacker.session.ended`` end-to-end.
+
+Wires :func:`_make_system_log_publisher` against a fake bus, drives
+two parsed events (a CMD then a session_recorded) through the
+returned publish_fn, and asserts the bus saw one
+``attacker.session.ended`` envelope alongside the per-line
+``system.log`` traffic.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncIterator
+
+import pytest
+import pytest_asyncio
+
+from decnet.bus import topics as _topics
+from decnet.bus.fake import FakeBus
+from decnet.collector.worker import _make_system_log_publisher
+
+
+@pytest_asyncio.fixture
+async def bus() -> AsyncIterator[FakeBus]:
+    b = FakeBus()
+    await b.connect()
+    try:
+        yield b
+    finally:
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_session_ended_published_alongside_system_log(
+    bus: FakeBus,
+) -> None:
+    captured: list[tuple[str, dict[str, Any]]] = []
+    sub = bus.subscribe(_topics.attacker(_topics.ATTACKER_SESSION_ENDED))
+
+    async def drain() -> None:
+        try:
+            async with sub:
+                async for ev in sub:
+                    captured.append((ev.topic, ev.payload))
+        except Exception:
+            pass
+
+    drain_task = asyncio.create_task(drain())
+    await asyncio.sleep(0)
+
+    loop = asyncio.get_running_loop()
+    publish_fn = _make_system_log_publisher(bus, loop)
+
+    publish_fn({
+        "timestamp": "2026-05-02T06:22:48",
+        "decky": "SRV-DELTA-77",
+        "service": "bash",
+        "event_type": "command",
+        "attacker_ip": "192.168.1.5",
+        "fields": {"command": "whoami"},
+    })
+    publish_fn({
+        "timestamp": "2026-05-02T06:23:00",
+        "decky": "omega-decky",
+        "service": "sessrec",
+        "event_type": "session_recorded",
+        "attacker_ip": "192.168.1.5",
+        "fields": {
+            "sid": "sess-abc",
+            "service": "ssh",
+            "duration_s": "60.0",
+        },
+    })
+
+    # Give the marshalled publish a tick to land.
+    await asyncio.sleep(0.1)
+    drain_task.cancel()
+
+    assert len(captured) == 1
+    topic, payload = captured[0]
+    assert topic == _topics.attacker(_topics.ATTACKER_SESSION_ENDED)
+    assert payload["session_id"] == "sess-abc"
+    assert [c["command_text"] for c in payload["commands"]] == ["whoami"]