feat: discovery-driven forensics hub + evidence board (RFC-13 Phase 6, #68)

forensics.investigate.hub runs the forensics pipeline as a discovery-driven
dispatch over the shared substrate, reusing every existing stage handler
unchanged. make_evidence_condition (new, content-aware) matches a
discovery's evidence_type, so a discovered disk image opens the disk and
binary lanes and a pcap opens the network lane, off the existing
_LANE_EVIDENCE_TYPES classification. Each phase adapter runs the real stage
and only overrides the transition back to the hub; each lane phase scopes
state_collection to its single lane via active_lanes. The deterministic
tail (deep_analysis, promotion, resolution, writeup) runs unconditionally
after the lanes.

record_evidence posts a discovered evidence item to the shared ledger as
the cross-branch evidence board (discovery entry carrying type, path,
source; idempotent per path).

No collector machinery is rewritten and the live FORENSICS_DISPATCHER_V1 is
untouched -- the hub ships bound nowhere, enabled by an operator seed
rebind after smoke, like the malware and vr hubs.

5 tests: hub structure (lanes + tail), disk_image activates disk, pcap
activates network, evidence board records provenance + is idempotent, and
no matching evidence falls through to the unconditional tail.

Author: echel0nn

CHANGELOG.md

@@ -89,6 +89,19 @@ operator action.
   -- so it activates only once the panel confirms an exploitable finding by
   quorum. Reuses the Phase 0-4 substrate with no new platform code; ships
   bound nowhere live (operator rebind after smoke).
+- Discovery-driven forensics hub + evidence board (RFC-13 #68,
+  `forensics.investigate.hub`): a content-aware `make_evidence_condition`
+  matches a discovery's `evidence_type`, so a discovered disk image opens
+  the disk and binary lanes and a discovered pcap opens the network lane,
+  reusing the existing `_LANE_EVIDENCE_TYPES` classification. The hub runs
+  the proven forensics stages unchanged -- each phase adapter runs the real
+  stage and only overrides the transition back to the hub, and each lane
+  phase scopes `state_collection` to its single lane via `active_lanes` --
+  then runs the deterministic tail (deep_analysis, promotion, resolution,
+  writeup) unconditionally. `record_evidence` posts a discovered evidence
+  item to the shared ledger as the cross-branch evidence board. No
+  collector machinery is rewritten and the live `FORENSICS_DISPATCHER_V1`
+  is untouched; the hub ships bound nowhere (operator rebind after smoke).
 - Platform agent runtime (RFC-03): `AgentTurnRunnerBase`,
   `ToolExecutorHelpersBase`, the shared turn helpers, and platform bases
   for the pattern extractor, claim verifier, synthesis runner, persona

src/aila/modules/forensics/workflow/definitions_hub.py

@@ -0,0 +1,169 @@
+"""Forensics dispatch-hub graph (opt-in) over the shared substrate.
+
+RFC-13 (#68). Where ``FORENSICS_DISPATCHER_V1`` routes to a linear mode
+pipeline, this graph runs the same proven stage handlers as a
+discovery-driven dispatch: setup (intake), then the hub, then a stage,
+then back to the hub, until emit. Intake enumerates evidence and posts the
+active lanes; the hub then activates a collector lane only when a matching
+evidence type was discovered -- a disk image opens the disk (and binary)
+lane, a pcap opens the network lane -- reusing the existing
+``_LANE_EVIDENCE_TYPES`` classification. The deterministic post-collection
+tail (deep_analysis, promotion, resolution, writeup) runs unconditionally
+after the lanes, in order.
+
+This reuses every existing forensics stage handler unchanged: each phase
+adapter runs the real stage and only overrides the next transition back to
+the hub, and each lane phase scopes ``state_collection`` to its single lane
+via ``active_lanes``. No collector machinery is rewritten and the live
+``FORENSICS_DISPATCHER_V1`` is untouched; the definition ships bound
+nowhere and is enabled by an operator seed rebind after a smoke.
+
+The shared ledger is the evidence board: :func:`record_evidence` posts a
+discovered evidence item as a ledger ``discovery`` with its type, path, and
+source so the hub conditions (and other branches) read a common board.
+"""
+from __future__ import annotations
+
+from typing import Any, cast
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from aila.modules.forensics.workflow.definitions import (
+    _build_services,
+    _state_response_emit,
+)
+from aila.modules.forensics.workflow.states.collection import (
+    _LANE_EVIDENCE_TYPES,
+    state_collection,
+)
+from aila.modules.forensics.workflow.states.deep_analysis import state_deep_analysis
+from aila.modules.forensics.workflow.states.intake import state_intake
+from aila.modules.forensics.workflow.states.promotion import state_promotion
+from aila.modules.forensics.workflow.states.resolution import state_resolution
+from aila.modules.forensics.workflow.states.writeup import state_writeup
+from aila.platform.services.ledger import LedgerService, make_evidence_condition
+from aila.platform.workflows.phase_graph import (
+    PhaseSpec,
+    build_dispatch_workflow,
+)
+from aila.platform.workflows.types import HandlerFn, StateResult
+
+__all__ = [
+    "FORENSICS_HUB_PHASES",
+    "FORENSICS_INVESTIGATE_HUB",
+    "record_evidence",
+]
+
+# Hub phase name -> collection lane key. The "binary" phase drives the
+# "binary_analysis" collection lane (capa/FLOSS/strings on samples the disk
+# lane surfaces).
+_PHASE_LANE: dict[str, str] = {
+    "disk": "disk",
+    "memory": "memory",
+    "network": "network",
+    "log": "log",
+    "binary": "binary_analysis",
+}
+
+# Hub phase name -> deterministic tail stage handler (unconditional).
+_TAIL_HANDLERS: dict[str, HandlerFn] = {
+    "deep_analysis": state_deep_analysis,
+    "promotion": state_promotion,
+    "resolution": state_resolution,
+    "writeup": state_writeup,
+}
+
+
+async def record_evidence(
+    investigation_id: str,
+    author_branch_id: str,
+    evidence_type: str,
+    path: str,
+    source: str,
+    *,
+    session: AsyncSession | None = None,
+) -> int:
+    """Post a discovered evidence item to the shared ledger (evidence board).
+
+    Recorded as a ``discovery`` entry carrying the evidence type, path, and
+    source so the hub's evidence conditions and other branches read one
+    board. Idempotency-keyed by path so re-enumerating the same evidence
+    does not double-post.
+    """
+    return await LedgerService().append_general(
+        investigation_id,
+        author_branch_id,
+        "discovery",
+        {"evidence_type": evidence_type, "path": path, "source": source},
+        idempotency_key=f"evidence:{path}",
+        session=session,
+    )
+
+
+def _setup_builder(next_state: str) -> HandlerFn:
+    """Run intake, then transition to the hub instead of a static edge."""
+    async def _handler(state_input: dict[str, Any], services: Any) -> StateResult:
+        result = await state_intake(state_input, services)
+        return StateResult(
+            next_state=next_state, output={**state_input, **result.output},
+        )
+    return _handler
+
+
+def _loop_builder(phase: PhaseSpec, next_state: str) -> HandlerFn:
+    """Run a lane collector or a tail stage, then loop back to the hub."""
+    lane = _PHASE_LANE.get(phase.name)
+    tail = _TAIL_HANDLERS.get(phase.name)
+
+    async def _handler(state_input: dict[str, Any], services: Any) -> StateResult:
+        if lane is not None:
+            scoped = {**state_input, "active_lanes": [lane]}
+            result = await state_collection(scoped, services)
+        else:
+            result = await tail(state_input, services)
+        output = result.output if isinstance(result, StateResult) else dict(result)
+        return StateResult(next_state=next_state, output={**state_input, **output})
+    return _handler
+
+
+FORENSICS_HUB_PHASES: tuple[PhaseSpec, ...] = (
+    PhaseSpec(
+        name="disk",
+        condition=make_evidence_condition(_LANE_EVIDENCE_TYPES["disk"]),
+        trust="advisory",
+    ),
+    PhaseSpec(
+        name="memory",
+        condition=make_evidence_condition(_LANE_EVIDENCE_TYPES["memory"]),
+        trust="advisory",
+    ),
+    PhaseSpec(
+        name="network",
+        condition=make_evidence_condition(_LANE_EVIDENCE_TYPES["network"]),
+        trust="advisory",
+    ),
+    PhaseSpec(
+        name="log",
+        condition=make_evidence_condition(_LANE_EVIDENCE_TYPES["log"]),
+        trust="advisory",
+    ),
+    PhaseSpec(
+        name="binary",
+        condition=make_evidence_condition(_LANE_EVIDENCE_TYPES["binary_analysis"]),
+        trust="advisory",
+    ),
+    PhaseSpec(name="deep_analysis"),
+    PhaseSpec(name="promotion"),
+    PhaseSpec(name="resolution"),
+    PhaseSpec(name="writeup"),
+)
+
+
+FORENSICS_INVESTIGATE_HUB = build_dispatch_workflow(
+    "forensics.investigate.hub",
+    FORENSICS_HUB_PHASES,
+    services_factory=_build_services,
+    setup_builder=_setup_builder,
+    loop_builder=_loop_builder,
+    emit_handler=cast("HandlerFn", _state_response_emit),
+)

src/aila/platform/services/ledger.py

@@ -20,7 +20,7 @@
 from __future__ import annotations
 
 import json
-from collections.abc import AsyncGenerator, Awaitable, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable
 from contextlib import asynccontextmanager
 from datetime import datetime
 from typing import Any
@@ -39,6 +39,7 @@
     "LedgerPermissionError",
     "LedgerService",
     "make_discovery_condition",
+    "make_evidence_condition",
 ]
 
 _UQ_IDEM = "uq_investigation_ledger_idem"
@@ -415,3 +416,44 @@ async def _condition(state_input: dict[str, Any]) -> tuple[bool, str]:
         return False, f"no {kind} entries on ledger yet"
 
     return _condition
+
+
+def make_evidence_condition(
+    evidence_types: str | Iterable[str],
+    *,
+    confirmed_only: bool = False,
+    input_key: str = "investigation_id",
+) -> Callable[[dict[str, Any]], Awaitable[tuple[bool, str]]]:
+    """Build a dispatch-hub condition that fires on a discovered evidence type.
+
+    A content-aware sibling of :func:`make_discovery_condition`: it reads
+    ``discovery`` entries and matches each entry's ``payload["evidence_type"]``
+    against *evidence_types*. The forensics hub uses it so a discovered disk
+    image activates the disk lane and a discovered pcap activates the network
+    lane, off the shared ledger (RFC-13 #68). ``confirmed_only`` restricts to
+    quorum-confirmed discoveries and honors the same ratified-replan relax
+    flag as :func:`make_discovery_condition`.
+    """
+    wanted = {evidence_types} if isinstance(evidence_types, str) else set(evidence_types)
+
+    async def _condition(state_input: dict[str, Any]) -> tuple[bool, str]:
+        investigation_id = state_input.get(input_key)
+        if not investigation_id:
+            return False, f"no {input_key} on dispatch input"
+        effective_confirmed = confirmed_only and not state_input.get(
+            "_dispatch_replan_relax"
+        )
+        entries = await LedgerService().read_general(
+            str(investigation_id),
+            kinds=["discovery"],
+            confirmed_only=effective_confirmed,
+        )
+        matched = [
+            e for e in entries
+            if (e.get("payload") or {}).get("evidence_type") in wanted
+        ]
+        if matched:
+            return True, f"{len(matched)} discovery entries matching {sorted(wanted)}"
+        return False, f"no discovery matching {sorted(wanted)}"
+
+    return _condition

tests/test_forensics_hub.py

@@ -0,0 +1,71 @@
+"""Forensics dispatch-hub: evidence-driven lane activation + evidence board (RFC-13 Phase 6)."""
+from __future__ import annotations
+
+from aila.modules.forensics.workflow.definitions_hub import (
+    FORENSICS_HUB_PHASES,
+    FORENSICS_INVESTIGATE_HUB,
+    record_evidence,
+)
+from aila.platform.services.ledger import LedgerService
+from aila.platform.workflows.phase_graph import DISPATCH_STATE, make_dispatch_router
+
+
+def test_hub_has_dispatch_lane_and_tail_states() -> None:
+    states = FORENSICS_INVESTIGATE_HUB.states
+    assert DISPATCH_STATE in states
+    for name in ("disk", "memory", "network", "log", "binary"):
+        assert name in states
+    for name in ("deep_analysis", "promotion", "resolution", "writeup"):
+        assert name in states
+    assert FORENSICS_INVESTIGATE_HUB.definition_id == "forensics.investigate.hub"
+
+
+async def test_disk_image_activates_disk_lane(test_db) -> None:
+    del test_db
+    inv = "inv-fx-disk"
+    await record_evidence(inv, "intake", "disk_image", "/ev/img.E01", "case-1")
+    router = make_dispatch_router(FORENSICS_HUB_PHASES)
+    result = await router({"investigation_id": inv, "_dispatch_visited": []}, None)
+    assert result.next_state == "disk"
+
+
+async def test_pcap_activates_network_lane(test_db) -> None:
+    del test_db
+    inv = "inv-fx-pcap"
+    await record_evidence(inv, "intake", "pcap", "/ev/capture.pcap", "case-2")
+    router = make_dispatch_router(FORENSICS_HUB_PHASES)
+    # Disk/memory lanes have no matching evidence, so the hub skips them and
+    # activates the network lane.
+    result = await router(
+        {"investigation_id": inv, "_dispatch_visited": ["disk", "memory"]}, None,
+    )
+    assert result.next_state == "network"
+
+
+async def test_evidence_board_records_provenance(test_db) -> None:
+    del test_db
+    inv = "inv-fx-board"
+    entry_id = await record_evidence(
+        inv, "intake", "memory_dump", "/ev/mem.raw", "case-3",
+    )
+    rows = await LedgerService().read_general(inv, kinds=["discovery"])
+    assert len(rows) == 1
+    payload = rows[0]["payload"]
+    assert rows[0]["id"] == entry_id
+    assert payload["evidence_type"] == "memory_dump"
+    assert payload["path"] == "/ev/mem.raw"
+    assert payload["source"] == "case-3"
+    # Re-recording the same path is idempotent (no double-post).
+    await record_evidence(inv, "intake", "memory_dump", "/ev/mem.raw", "case-3")
+    rows_again = await LedgerService().read_general(inv, kinds=["discovery"])
+    assert len(rows_again) == 1
+
+
+async def test_no_matching_evidence_falls_through_to_tail(test_db) -> None:
+    del test_db
+    inv = "inv-fx-tail"
+    # No evidence discovered: every lane condition is false, so the hub
+    # activates the first unconditional tail stage.
+    router = make_dispatch_router(FORENSICS_HUB_PHASES)
+    result = await router({"investigation_id": inv, "_dispatch_visited": []}, None)
+    assert result.next_state == "deep_analysis"