Queryable firewall/flow claims + cross-server query surface (CLI + MCP)

exports/manifest.json

@@ -1 +1 @@
-{"claim_count":5660,"claim_extracted_at":"2026-07-10T20:13:21Z","concept_count":918,"context_pack_count":0,"edge_count":1981,"enrichment_run_count":2,"eval_case_count":53,"eval_result_count":53,"grounding_row_count":1,"grounding_version":"grounding_v1","insight_decision_count":4,"insight_label_count":4,"insight_policy_version":"agent_core_insight_v1","learning_event_count":4,"learning_ledger_version":"learning_ledger_v1","learning_review_count":0,"loop_decision_envelope_count":0,"loop_decision_memory_version":"agent_core_loop_decision_v1","observation_count":2,"policy_decision_count":0,"policy_version":"knowledge_policy_v1","quality_finding_count":44,"retrieval_version":"retrieval_v1","run_id":"local-20260710201321","source_ref_count":1068,"source_shas":{}}
+{"claim_count":5660,"claim_extracted_at":"2026-07-11T13:12:51Z","concept_count":918,"context_pack_count":0,"edge_count":1981,"enrichment_run_count":2,"eval_case_count":53,"eval_result_count":53,"grounding_row_count":1,"grounding_version":"grounding_v1","insight_decision_count":4,"insight_label_count":4,"insight_policy_version":"agent_core_insight_v1","learning_event_count":4,"learning_ledger_version":"learning_ledger_v1","learning_review_count":0,"loop_decision_envelope_count":0,"loop_decision_memory_version":"agent_core_loop_decision_v1","observation_count":2,"policy_decision_count":0,"policy_version":"knowledge_policy_v1","quality_finding_count":44,"retrieval_version":"retrieval_v1","run_id":"local-20260711131251","source_ref_count":1068,"source_shas":{}}

knowledge-policy.yml

@@ -18,6 +18,10 @@ roles:
       - knowledge.search
       - knowledge.claims
       - knowledge.neighborhood
+      - knowledge.firewall_rules
+      - knowledge.server_registry
+      - knowledge.flows_into
+      - knowledge.flows_from
       - knowledge.context_pack.generate
       - knowledge.intended_state.read
       - knowledge.observed_state.read
@@ -47,6 +51,10 @@ roles:
       - knowledge.search
       - knowledge.claims
       - knowledge.neighborhood
+      - knowledge.firewall_rules
+      - knowledge.server_registry
+      - knowledge.flows_into
+      - knowledge.flows_from
       - knowledge.context_pack.generate
       - knowledge.intended_state.read
       - knowledge.observed_state.read
@@ -66,6 +74,10 @@ roles:
       - knowledge.resolve
       - knowledge.search
       - knowledge.claims
+      - knowledge.firewall_rules
+      - knowledge.server_registry
+      - knowledge.flows_into
+      - knowledge.flows_from
       - knowledge.context_pack.generate
       - knowledge.intended_state
       - knowledge.endpoint_schema

src/hyrule_knowledge/builder.py

@@ -10,7 +10,7 @@
 from typing import Any
 
 from .config import AppConfig, SourceConfig
-from .extractors.ansible import HostInfo, extract_hosts, extract_network_constants
+from .extractors.ansible import FlowInfo, HostInfo, extract_hosts, extract_network_constants
 from .extractors.api import ApiEndpointInfo, PydanticModelInfo, extract_api, model_concept_id
 from .extractors.dns import DnsZoneInfo, extract_zones
 from .extractors.monitoring import monitoring_check_names, monitoring_config_paths
@@ -948,30 +948,66 @@ def build_static_deployment_concept(
     )
 
 
+def _flow_bullet(flow: FlowInfo) -> str:
+    purpose = flow.purpose or "no purpose recorded"
+    if flow.direction == "outbound":
+        return f"`{flow.proto}` port `{flow.port}` to `{flow.peer.label}` — {purpose}"
+    return f"`{flow.proto}` port `{flow.port}` from `{flow.peer.label}` — {purpose}"
+
+
+def host_connect_targets(host: HostInfo, host_names: set[str]) -> list[str]:
+    """Concept ids this host connects out to (for ``connects-to`` edges).
+
+    Only outbound flows whose peer resolves to another known host produce an
+    edge; wildcard/external/subnet peers and self-loops are skipped. Sorted and
+    de-duplicated for deterministic edge emission.
+    """
+    self_id = _host_concept_id(host.name)
+    targets: set[str] = set()
+    for flow in host.flows:
+        if flow.direction != "outbound" or not flow.peer.host or flow.peer.host not in host_names:
+            continue
+        target_id = _host_concept_id(flow.peer.host)
+        if target_id != self_id:
+            targets.add(target_id)
+    return sorted(targets)
+
+
 def build_host_concept(snapshot: RepoSnapshot, host: HostInfo, verified_at: str) -> Concept:
     concept_id = _host_concept_id(host.name)
     firewall = [
         f"`{rule.proto or 'any'}` port `{rule.dport or 'any'}` from `{rule.src}` — {rule.comment or 'no comment'}"
         for rule in host.firewall_rules
     ]
+    outbound = [_flow_bullet(flow) for flow in host.flows if flow.origin == "outbound"]
+    cross_cutting = [_flow_bullet(flow) for flow in host.flows if flow.origin == "cross-cutting"]
     monitoring = [
         f"`{service.name}` ({service.check_command or 'unspecified'}) — {service.notes or 'no notes'}"
         for service in host.monitoring_services
     ]
     pins = [f"`{key}` = `{value}`" for key, value in host.version_pins.items()]
     peer_rows = _yaml_table_rows(host.peer)
-    host_var_rows = _yaml_table_rows({k: v for k, v in host.host_vars.items() if k not in {"firewall_extra_rules", "monitoring_extra_services"}})
+    host_var_rows = _yaml_table_rows(
+        {
+            k: v
+            for k, v in host.host_vars.items()
+            if k not in {"firewall_extra_rules", "monitoring_extra_services", "host_meta", "network_flows_outbound"}
+        }
+    )
     body = f"""# Host
 
 | Field | Value |
 | --- | --- |
 | Host | `{host.name}` |
 | Type | `{host.concept_type}` |
+| OS | `{host.os or 'not specified'}` |
+| Role | `{host.role or 'not specified'}` |
 | Address | `{host.address or 'not specified'}` |
 | Groups | `{', '.join(host.groups)}` |
 | Role comment | `{host.role_comment or 'not found'}` |
 | Monitoring role | `{host.monitoring_role or 'not set'}` |
 | Logs role | `{host.logs_role or 'not set'}` |
+| Summary | {host.summary or 'not specified'} |
 
 # Deployed version pins
 
@@ -981,6 +1017,14 @@ def build_host_concept(snapshot: RepoSnapshot, host: HostInfo, verified_at: str)
 
 {_markdown_list(firewall, 'No `firewall_extra_rules` found in host vars.')}
 
+# Outbound flows
+
+{_markdown_list(outbound, 'No `network_flows_outbound` found in host vars.')}
+
+# Cross-cutting flows
+
+{_markdown_list(cross_cutting, 'No cross-cutting flows reference this host.')}
+
 # Monitoring services
 
 {_markdown_list(monitoring, 'No host-local `monitoring_extra_services` found in host vars.')}
@@ -1004,6 +1048,19 @@ def build_host_concept(snapshot: RepoSnapshot, host: HostInfo, verified_at: str)
     if (snapshot.path / host_var_path).exists():
         refs.append(source_ref(snapshot, host_var_path))
         body += f"[2] [{host_var_path}]({source_url(snapshot.url, snapshot.commit, host_var_path)})\n"
+    flows_path = "ansible/inventory/network_flows.yml"
+    if (snapshot.path / flows_path).exists():
+        refs.append(source_ref(snapshot, flows_path))
+        body += f"[{len(refs)}] [{flows_path}]({source_url(snapshot.url, snapshot.commit, flows_path)})\n"
+    extra: dict[str, Any] = {"repo": snapshot.repo, "host": host.name, "address": host.address, "groups": host.groups}
+    if host.os:
+        extra["os"] = host.os
+    if host.role:
+        extra["role"] = host.role
+    if host.summary:
+        extra["summary"] = host.summary
+    if host.flows:
+        extra["network_flows"] = [flow.as_dict() for flow in host.flows]
     return Concept(
         concept_id=concept_id,
         concept_type=host.concept_type,
@@ -1018,7 +1075,7 @@ def build_host_concept(snapshot: RepoSnapshot, host: HostInfo, verified_at: str)
         last_verified_at=verified_at,
         confidence="high",
         dispute_policy="repo_wins",
-        extra={"repo": snapshot.repo, "host": host.name, "address": host.address, "groups": host.groups},
+        extra=extra,
         body=body,
     )
 
@@ -1406,11 +1463,14 @@ def build_all(
                     result.edge(endpoint_concept.concept_id, schema_id, "uses-schema")
 
         if snapshot.name == "network-operations":
+            host_names = {host.name for host in context.hosts}
             for host in context.hosts:
                 host_concept = build_host_concept(snapshot, host, verified_at)
                 result.add(host_concept)
                 if project_concept:
                     result.edge(project_concept.concept_id, host_concept.concept_id, "owns-intended-state")
+                for target_id in host_connect_targets(host, host_names):
+                    result.edge(host_concept.concept_id, target_id, "connects-to")
                 for pin, value in host.version_pins.items():
                     service_repo = _repo_for_pin(pin)
                     service_context = context_by_repo.get(service_repo) if service_repo else None

src/hyrule_knowledge/claims.py

@@ -93,6 +93,33 @@ def _frontmatter(concept: dict[str, Any]) -> dict[str, Any]:
     return raw if isinstance(raw, dict) else {}
 
 
+def _section(body: str, header: str) -> str:
+    """The body text under a ``# <header>`` heading, up to the next ``# ``
+    heading. Used to scope the legacy firewall regex to just its own section."""
+    out: list[str] = []
+    capturing = False
+    for line in body.splitlines():
+        if line.strip() == header:
+            capturing = True
+            continue
+        if capturing and line.startswith("# "):
+            break
+        if capturing:
+            out.append(line)
+    return "\n".join(out)
+
+
+def _ref_index(concept: dict[str, Any], suffix: str) -> int | None:
+    """1-based index of the source_ref whose path ends with ``suffix`` (so a
+    claim can cite the file the fact actually came from), or None."""
+    refs = concept.get("source_refs")
+    if isinstance(refs, list):
+        for i, ref in enumerate(refs, start=1):
+            if isinstance(ref, dict) and str(ref.get("path") or "").endswith(suffix):
+                return i
+    return None
+
+
 def _source_uri(concept: dict[str, Any], index: int = 1) -> str:
     refs_obj = concept.get("source_refs")
     refs: list[Any] = refs_obj if isinstance(refs_obj, list) else []
@@ -189,22 +216,122 @@ def _host_claims(concept: dict[str, Any], extracted_at: str) -> list[KnowledgeCl
     claims: list[KnowledgeClaim] = []
     if not host:
         return claims
+    subject = f"host:{host}"
+    # host_meta (os/role) lives in host_vars/<host>.yml — cite it, not hosts.yml.
+    host_vars_ref = _ref_index(concept, f"host_vars/{host}.yml") or 1
+    flows_ref = _ref_index(concept, "network_flows.yml") or host_vars_ref
     address = fm.get("address")
     if address:
-        claims.append(_claim(concept, subject=f"host:{host}", predicate="has_address", object_value=str(address), extracted_at=extracted_at))
+        claims.append(_claim(concept, subject=subject, predicate="has_address", object_value=str(address), extracted_at=extracted_at))
     groups = fm.get("groups")
     if isinstance(groups, list):
         for group in groups:
-            claims.append(_claim(concept, subject=f"host:{host}", predicate="member_of_group", object_value=str(group), extracted_at=extracted_at))
+            claims.append(_claim(concept, subject=subject, predicate="member_of_group", object_value=str(group), extracted_at=extracted_at))
+    os_value = str(fm.get("os") or "").strip()
+    if os_value:
+        claims.append(_claim(concept, subject=subject, predicate="runs_os", object_value=os_value, extracted_at=extracted_at, source_ref_index=host_vars_ref))
+    role_value = str(fm.get("role") or "").strip()
+    if role_value:
+        claims.append(_claim(concept, subject=subject, predicate="has_role", object_value=role_value, extracted_at=extracted_at, source_ref_index=host_vars_ref))
     body = str(concept.get("body") or "")
-    for role_name, role_value in _HOST_ROLE_RE.findall(body):
-        if role_value in {"not set", "not found"}:
+    for role_name, role_val in _HOST_ROLE_RE.findall(body):
+        if role_val in {"not set", "not found"}:
             continue
         predicate = "has_monitoring_role" if role_name == "Monitoring role" else "has_logs_role"
-        claims.append(_claim(concept, subject=f"host:{host}", predicate=predicate, object_value=role_value, extracted_at=extracted_at))
-    for match in _FIREWALL_RE.finditer(body):
+        claims.append(_claim(concept, subject=subject, predicate=predicate, object_value=role_val, extracted_at=extracted_at))
+    # `allows_inbound` is the firewall_extra_rules projection ONLY — scan just
+    # the "# Inbound firewall rules" section so cross-cutting/outbound flow
+    # bullets (same `port ... from` format) don't leak in as phantom rules.
+    for match in _FIREWALL_RE.finditer(_section(body, "# Inbound firewall rules")):
         value = f"{match.group('proto')}/{match.group('port')}/{match.group('src')}"
-        claims.append(_claim(concept, subject=f"host:{host}", predicate="allows_inbound", object_value=value, extracted_at=extracted_at))
+        claims.append(_claim(concept, subject=subject, predicate="allows_inbound", object_value=value, extracted_at=extracted_at, source_ref_index=host_vars_ref))
+    claims.extend(_flow_claims(concept, subject, fm, extracted_at, host_vars_ref=host_vars_ref, flows_ref=flows_ref))
+    return claims
+
+
+def _flow_claims(
+    concept: dict[str, Any],
+    subject: str,
+    fm: dict[str, Any],
+    extracted_at: str,
+    *,
+    host_vars_ref: int = 1,
+    flows_ref: int = 1,
+) -> list[KnowledgeClaim]:
+    """Decomposed, cross-server-queryable claims from resolved host flows.
+
+    Reads the resolved ``network_flows`` frontmatter block written by
+    ``build_host_concept`` (each entry carries the fully-resolved peer host,
+    address, and — where ``all`` was narrowed — the excluded hosts). Emits, per
+    direction:
+
+    * inbound: ``allows_port`` / ``allows_proto`` (bare-value index claims) and
+      one aggregated ``allows_from`` per source peer.
+    * outbound: ``connects_port`` / ``connects_proto`` and one aggregated
+      ``connects_to`` per destination peer.
+    """
+    raw = fm.get("network_flows")
+    if not isinstance(raw, list):
+        return []
+    flows = [entry for entry in raw if isinstance(entry, dict)]
+    claims: list[KnowledgeClaim] = []
+    directions: dict[str, list[dict[str, Any]]] = {"inbound": [], "outbound": []}
+    for entry in flows:
+        direction = str(entry.get("direction") or "")
+        if direction in directions:
+            directions[direction].append(entry)
+
+    def _ref_for(entries: list[dict[str, Any]]) -> int:
+        # Cite network_flows.yml only when every flow in the group is a
+        # cross-cutting flow; otherwise the fact came from the host vars.
+        origins = {str(e.get("origin") or "") for e in entries}
+        return flows_ref if origins == {"cross-cutting"} else host_vars_ref
+
+    config = {
+        "inbound": ("allows_port", "allows_proto", "allows_from", "src_host", "src_addr", "src_kind"),
+        "outbound": ("connects_port", "connects_proto", "connects_to", "dst_host", "dst_addr", "dst_kind"),
+    }
+    for direction, entries in directions.items():
+        port_pred, proto_pred, rel_pred, host_key, addr_key, kind_key = config[direction]
+        ports = sorted({str(entry.get("port")) for entry in entries if entry.get("port") is not None})
+        protos = sorted({str(entry.get("proto")) for entry in entries if entry.get("proto") is not None})
+        for port in ports:
+            claims.append(_claim(concept, subject=subject, predicate=port_pred, object_value=port, extracted_at=extracted_at, source_ref_index=host_vars_ref, metadata={"direction": direction}))
+        for proto in protos:
+            claims.append(_claim(concept, subject=subject, predicate=proto_pred, object_value=proto, extracted_at=extracted_at, source_ref_index=host_vars_ref, metadata={"direction": direction}))
+        # One aggregated claim per peer (object). The claim id hashes only
+        # subject/predicate/object/source_uri (not metadata), so grouping by
+        # peer_addr would mint two claims with the SAME id for a peer reachable
+        # on >1 address (router loopback + underlay) and `compile_claims` would
+        # drop the second. Instead keep one claim and preserve every address in
+        # the per-flow `addr` field so no address is lost.
+        grouped: dict[tuple[str, str], list[dict[str, Any]]] = {}
+        for entry in entries:
+            peer_host = entry.get("peer_host")
+            object_value = f"host:{peer_host}" if peer_host else str(entry.get("peer") or "unknown")
+            grouped.setdefault((object_value, str(peer_host or "")), []).append(entry)
+        for (object_value, _peer_host), grouped_entries in sorted(grouped.items()):
+            sample = grouped_entries[0]
+            peer_host = sample.get("peer_host")
+            flow_rows = sorted(
+                ({"port": str(entry.get("port")), "proto": str(entry.get("proto")), "family": str(entry.get("family") or ""), "addr": str(entry.get("peer_addr") or ""), "purpose": str(entry.get("purpose") or ""), "origin": str(entry.get("origin") or "")} for entry in grouped_entries),
+                key=lambda row: (row["proto"], row["port"], row["family"], row["addr"], row["origin"], row["purpose"]),
+            )
+            addresses = sorted({str(entry.get("peer_addr")) for entry in grouped_entries if entry.get("peer_addr")})
+            metadata: dict[str, Any] = {
+                "direction": direction,
+                host_key: peer_host,
+                addr_key: sample.get("peer_addr"),
+                kind_key: sample.get("peer_kind"),
+                "peer": sample.get("peer"),
+                "flows": flow_rows,
+            }
+            if len(addresses) > 1:
+                metadata[f"{addr_key}s"] = addresses
+            excludes = sample.get("excludes")
+            if isinstance(excludes, list) and excludes:
+                metadata["excludes"] = list(excludes)
+            claims.append(_claim(concept, subject=subject, predicate=rel_pred, object_value=object_value, extracted_at=extracted_at, source_ref_index=_ref_for(grouped_entries), metadata=metadata))
     return claims