feat: planner oracle + discovery-driven malware hub (RFC-13 Phase 4, #68)

The core loop. Oracle (platform/services/oracle.py) is a thin request
router over the shared ledger: route_pending resolves each open request
to its decider by target_capability (the objective owner, or a quorum),
record_decision enforces a distinct-approver rule so a branch cannot
ratify its own request, and apply_decision applies ONLY the declared
mechanical effect once ratified -- confirm the discovery (activate_phase),
open or write an objective, or nothing (replan). It decides nothing and
runs no LLM planner.

The dispatch hub gains: overall-cap handling (emits budget_truncated on
_budget_exhausted), a stall path that raises one replan request per
distinct visited-set when unvisited phases are blocked by unmet
conditions, and a ratified-replan relaxation -- a make_discovery_condition
reads _dispatch_replan_relax and drops confirmed trust to advisory for one
pass, so a quorum deadlock cannot freeze the graph. The hub's stall writer
uses lazy service imports (phase_graph loads during db_models init), with
a documented PLC0415 per-file-ignore.

malware.investigate.hub runs the malware phases as a discovery-driven
dispatch: triage (readiness), unpack (capability=re), config_extract
(capability=crypto) both confirmed-trust, and full_analysis the
unconditional advisory fallback. It ships bound nowhere live -- enabled by
an operator seed rebind after smoke, exactly like the V2 graphs; the live
V1 path is untouched.

14 tests: oracle routing/apply/reject/self-approval/quorum-k; the hub run
through DurableStateMachine (setup->hub->A->hub->B->emit, budget cut,
stall raises replan, resume preserves the visited-set); malware hub
structure + capability/trust wiring + confirmed-discovery activation.

Author: echel0nn

CHANGELOG.md

@@ -66,6 +66,21 @@ operator action.
   closed branch's objectives to the investigation, so a terminal branch
   never keeps a live objective. No new table (objectives stay tagged
   ledger entries).
+- Planner oracle + discovery-driven malware hub (RFC-13 #68): a thin
+  request router (`platform/services/oracle.py`) resolves ledger requests
+  (activate_phase / open_objective / write_objective / replan) to their
+  decider by `target_capability`, enforces a distinct-approver rule (a
+  branch cannot ratify its own request), and applies only the declared
+  mechanical effect once a quorum ratifies -- it decides nothing itself.
+  The dispatch hub gained overall-cap handling (emits `budget_truncated`),
+  a stall path that raises one `replan` request per visited-set, and a
+  ratified-replan relaxation that drops confirmed trust to advisory for
+  one pass so a quorum deadlock cannot freeze the graph.
+  `malware.investigate.hub` ships the malware phases as a discovery-driven
+  dispatch (unpack capability=re, config_extract capability=crypto, both
+  confirmed-trust; full_analysis the advisory fallback), bound nowhere
+  live -- enabled by an operator seed rebind after smoke, like the V2
+  graphs.
 - Platform agent runtime (RFC-03): `AgentTurnRunnerBase`,
   `ToolExecutorHelpersBase`, the shared turn helpers, and platform bases
   for the pattern extractor, claim verifier, synthesis runner, persona

pyproject.toml

@@ -408,6 +408,11 @@ ignore = [
 "src/aila/platform/workflows/engine.py" = ["BLE001", "PLC0415"]
 "src/aila/platform/workflows/errors.py" = ["N818"]
 "src/aila/platform/workflows/log.py" = ["PLC0415", "BLE001"]
+# phase_graph.py: PLC0415 for the deferred LedgerService / UnitOfWork imports
+# inside the dispatch hub's stall handler. phase_graph is imported while
+# db_models is still loading (tasks -> workflows -> phase_graph), so a
+# module-scope services import would re-enter a half-built db_models.
+"src/aila/platform/workflows/phase_graph.py" = ["PLC0415"]
 "src/aila/storage/database.py" = ["PLC0415"]
 "src/aila/storage/db_models.py" = ["E402", "F401"]
 "src/aila/storage/operations.py" = ["PLC0415"]

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

@@ -0,0 +1,98 @@
+"""Malware investigation dispatch-hub graph (opt-in) over the shared substrate.
+
+RFC-13 (#68). Where ``malware.investigate.v2`` wires the phases by static
+edges and a kind router, this graph runs them as a discovery-driven
+dispatch: setup, then the hub, then a phase, then back to the hub, until
+emit. The hub activates a phase only when its ``condition`` holds, so the
+deep phases fire on real discoveries posted to the shared ledger rather
+than on a fixed edge.
+
+Phase activation:
+
+* ``triage`` -- gated on target readiness (the same static-analysis gate
+  V2 uses). It runs first and posts its structural findings to the ledger.
+* ``unpack`` -- ``capability="re"``, ``trust="confirmed"``: activates once
+  a discovery is confirmed by a quorum decision (the reverse-engineering
+  specialist's phase).
+* ``config_extract`` -- ``capability="crypto"``, ``trust="confirmed"``:
+  activates on a confirmed discovery (the config/crypto specialist).
+* ``full_analysis`` -- unconditional, ``trust="advisory"``, declared last:
+  the fallback deep phase that always remains available so the hub never
+  dead-ends when no discovery routed to a specialist phase.
+
+Malware runs a single agent server (ida_headless), so ``_branch_capability``
+is not threaded and the capability fields document intent rather than
+enforce per-branch routing (that matters for the multi-persona vr hub).
+The definition ships alongside V1 and V2; a seed binds it per investigation
+only after an operator smoke.
+"""
+from __future__ import annotations
+
+from typing import cast
+
+from aila.modules.malware.workflow.definitions import _build_services
+from aila.modules.malware.workflow.definitions_v2 import (
+    _CONFIG_EXTRACT_DIRECTIVE,
+    _FULL_ANALYSIS_DIRECTIVE,
+    _TRIAGE_DIRECTIVE,
+    _loop_builder,
+    _setup_builder,
+    _target_ready,
+)
+from aila.modules.malware.workflow.states.investigation_emit import (
+    state_investigation_emit,
+)
+from aila.platform.services.ledger import make_discovery_condition
+from aila.platform.workflows.phase_graph import (
+    PhaseSpec,
+    build_dispatch_workflow,
+)
+from aila.platform.workflows.types import HandlerFn
+
+__all__ = ["MALWARE_HUB_PHASES", "MALWARE_INVESTIGATE_HUB"]
+
+_UNPACK_DIRECTIVE = (
+    "UNPACKING PHASE. Objective: defeat the packer or obfuscation blocking "
+    "static analysis -- identify the packer, dump the unpacked image, and "
+    "reconstruct imports so the deeper phases read real code. Submit once "
+    "the sample is statically analyzable."
+)
+
+MALWARE_HUB_PHASES: tuple[PhaseSpec, ...] = (
+    PhaseSpec(
+        name="triage",
+        directive=_TRIAGE_DIRECTIVE,
+        max_turns=20,
+        condition=_target_ready,
+        trust="confirmed",
+    ),
+    PhaseSpec(
+        name="unpack",
+        directive=_UNPACK_DIRECTIVE,
+        condition=make_discovery_condition("discovery", confirmed_only=True),
+        capability="re",
+        trust="confirmed",
+    ),
+    PhaseSpec(
+        name="config_extract",
+        directive=_CONFIG_EXTRACT_DIRECTIVE,
+        condition=make_discovery_condition("discovery", confirmed_only=True),
+        capability="crypto",
+        trust="confirmed",
+    ),
+    PhaseSpec(
+        name="full_analysis",
+        directive=_FULL_ANALYSIS_DIRECTIVE,
+        trust="advisory",
+    ),
+)
+
+
+MALWARE_INVESTIGATE_HUB = build_dispatch_workflow(
+    "malware.investigate.hub",
+    MALWARE_HUB_PHASES,
+    services_factory=_build_services,
+    setup_builder=_setup_builder,
+    loop_builder=_loop_builder,
+    emit_handler=cast("HandlerFn", state_investigation_emit),
+)

src/aila/platform/services/__init__.py

@@ -8,6 +8,7 @@
 from .http import build_http_client
 from .knowledge import KnowledgeService
 from .ledger import LedgerService
+from .oracle import Oracle
 from .reasoning import CyberReasoningEngine
 from .reasoning_graphs import ReasoningGraphService
 from .report import ReportService
@@ -22,6 +23,7 @@
     "KnowledgeService",
     "LedgerService",
     "MiniLMProvider",
+    "Oracle",
     "ReasoningGraphService",
     "ReportService",
     "SSHService",

src/aila/platform/services/ledger.py

@@ -398,13 +398,19 @@ 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"
+        # A ratified replan relaxes confirmed trust for one hub pass, so a
+        # confirmed-trust phase can activate on an unconfirmed discovery
+        # rather than deadlocking when quorum never confirms (RFC-13 #68).
+        effective_confirmed = confirmed_only and not state_input.get(
+            "_dispatch_replan_relax"
+        )
         entries = await LedgerService().read_general(
             str(investigation_id),
             kinds=[kind],
-            confirmed_only=confirmed_only,
+            confirmed_only=effective_confirmed,
         )
         if entries:
-            scope = "confirmed " if confirmed_only else ""
+            scope = "confirmed " if effective_confirmed else ""
             return True, f"{len(entries)} {scope}{kind} entries on ledger"
         return False, f"no {kind} entries on ledger yet"