From 09d3a2dd5a36a174c26d17cc92ae3122a4d98800 Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 11:31:51 +0300
Subject: [PATCH 1/6] feat(node): node role/branch config seam + fail-fast boot
 (F2 step 1)

Three-node demo topology (ADR 0012). New src/serving/node/ resolves and
validates AGENTFLOW_NODE_* once at boot: unset/empty role == standalone
(byte-identical to today's demo, N1); center/edge fully validated with
fail-fast on a missing center URL, missing token, or unknown branch.
Lifespan stores node_config/role/branch on app.state; the ingest endpoint
and edge emitter (later steps) hang off this single seam.

N1/N2 config + standalone-unchanged tests (unit + integration).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/serving/api/main.py                 |   8 +
 src/serving/node/__init__.py            |  30 ++++
 src/serving/node/config.py              | 127 +++++++++++++++
 tests/integration/test_node_topology.py |  65 ++++++++
 tests/unit/test_node_config.py          | 197 ++++++++++++++++++++++++
 5 files changed, 427 insertions(+)
 create mode 100644 src/serving/node/__init__.py
 create mode 100644 src/serving/node/config.py
 create mode 100644 tests/integration/test_node_topology.py
 create mode 100644 tests/unit/test_node_config.py

diff --git a/src/serving/api/main.py b/src/serving/api/main.py
index 666802d..3480352 100644
--- a/src/serving/api/main.py
+++ b/src/serving/api/main.py
@@ -56,6 +56,7 @@
 from src.serving.cache import QueryCache
 from src.serving.control_plane import control_plane_store_kind, get_control_plane_store
 from src.serving.db_pool import DuckDBPool
+from src.serving.node import resolve_node_config
 from src.serving.semantic_layer.catalog import DataCatalog
 from src.serving.semantic_layer.query_engine import QueryEngine
 from src.serving.semantic_layer.search_index import SearchIndex
@@ -85,6 +86,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     setup_telemetry(app)
     app.state.demo_mode = os.getenv("AGENTFLOW_DEMO_MODE", "").lower() == "true"
     app.state.demo_seed_on_boot = os.getenv("AGENTFLOW_SEED_ON_BOOT", "").lower() == "true"
+    # Three-node demo topology (ADR 0012): resolve role/branch/token once here
+    # and fail fast on a misconfigured node. Unset role == standalone, which is
+    # byte-identical to today's single-node demo (N1). The center ingest
+    # endpoint and the edge emitter hang off this resolved config.
+    app.state.node_config = resolve_node_config()
+    app.state.node_role = app.state.node_config.role
+    app.state.node_branch = app.state.node_config.branch
     # Reset the auth-disabled bypass flag on every lifespan startup. This is a
     # process-wide attribute and tests may toggle it; without an explicit
     # reset a later TestClient lifespan with no configured keys would silently
diff --git a/src/serving/node/__init__.py b/src/serving/node/__init__.py
new file mode 100644
index 0000000..4c592f6
--- /dev/null
+++ b/src/serving/node/__init__.py
@@ -0,0 +1,30 @@
+"""Three-node demo topology (ADR 0012) — node role/branch seam.
+
+One image, three roles (``center`` / ``edge`` / ``standalone``) differentiated
+purely by environment. This package resolves and validates that environment
+once at boot; the center ingest endpoint, the edge emitter, and the
+branch-scoped seed all read the resolved :class:`NodeConfig` from
+``app.state`` rather than re-reading ``os.environ``.
+
+See ``docs/three-node-demo-topology.md`` for the build contract.
+"""
+
+from src.serving.node.config import (
+    CENTER_BRANCH,
+    EDGE_BRANCHES,
+    KNOWN_BRANCHES,
+    NodeConfig,
+    NodeConfigError,
+    NodeRole,
+    resolve_node_config,
+)
+
+__all__ = [
+    "CENTER_BRANCH",
+    "EDGE_BRANCHES",
+    "KNOWN_BRANCHES",
+    "NodeConfig",
+    "NodeConfigError",
+    "NodeRole",
+    "resolve_node_config",
+]
diff --git a/src/serving/node/config.py b/src/serving/node/config.py
new file mode 100644
index 0000000..698a385
--- /dev/null
+++ b/src/serving/node/config.py
@@ -0,0 +1,127 @@
+"""Resolve and validate the ``AGENTFLOW_NODE_*`` environment (ADR 0012 §2).
+
+The node role is pure environment on top of the existing demo image. This
+module is the single place that reads that environment, so a misconfigured
+node fails fast at boot (``docs/three-node-demo-topology.md`` §2: "boot must
+fail fast if an edge has role=edge but no center URL or no token") instead of
+coming up half-wired.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Literal
+
+NodeRole = Literal["center", "edge", "standalone"]
+
+# Legend mapping (ADR 0012 Decision 1): center = HQ ``msk``; the two live edge
+# nodes are the RU regional warehouses ``spb`` / ``ekb``. Foreign branches
+# (``dxb`` / ``ala``) are deliberately NOT live nodes (PII border), so they are
+# absent here.
+CENTER_BRANCH = "msk"
+EDGE_BRANCHES = frozenset({"spb", "ekb"})
+KNOWN_BRANCHES = frozenset({CENTER_BRANCH}) | EDGE_BRANCHES
+
+_VALID_ROLES = frozenset({"center", "edge", "standalone"})
+
+
+class NodeConfigError(ValueError):
+    """The ``AGENTFLOW_NODE_*`` environment is internally inconsistent.
+
+    Raised at boot so a misconfigured node fails fast rather than serving a
+    half-wired surface: an edge with no center URL to emit to, a node with no
+    token to authenticate ingest, or an unknown branch.
+    """
+
+
+@dataclass(frozen=True)
+class NodeConfig:
+    """Resolved node identity. ``standalone`` carries no branch/url/token —
+    it is today's single-node demo, unchanged (a strict superset, N1)."""
+
+    role: NodeRole
+    branch: str | None = None
+    center_url: str | None = None
+    token: str | None = None
+
+    @property
+    def is_center(self) -> bool:
+        return self.role == "center"
+
+    @property
+    def is_edge(self) -> bool:
+        return self.role == "edge"
+
+    @property
+    def is_standalone(self) -> bool:
+        return self.role == "standalone"
+
+
+def _clean(value: str | None) -> str | None:
+    """Trim to ``None`` when unset or whitespace-only."""
+    if value is None:
+        return None
+    trimmed = value.strip()
+    return trimmed or None
+
+
+def resolve_node_config(env: Mapping[str, str] | None = None) -> NodeConfig:
+    """Resolve the node role/branch/center-url/token from the environment.
+
+    ``AGENTFLOW_NODE_ROLE`` unset or ``standalone`` yields the unchanged
+    single-node demo. ``center`` and ``edge`` are fully validated; any
+    inconsistency raises :class:`NodeConfigError`.
+    """
+    if env is None:
+        env = os.environ
+
+    raw_role = (env.get("AGENTFLOW_NODE_ROLE") or "").strip().lower()
+    role = raw_role or "standalone"
+    if role not in _VALID_ROLES:
+        raise NodeConfigError(
+            f"AGENTFLOW_NODE_ROLE={raw_role!r} is not one of "
+            f"{sorted(_VALID_ROLES)} (unset or empty means standalone)."
+        )
+
+    if role == "standalone":
+        # Strict superset: stray branch/url/token vars are ignored, never
+        # validated, so the current demo Space keeps booting with no new env.
+        return NodeConfig(role="standalone")
+
+    branch = (_clean(env.get("AGENTFLOW_NODE_BRANCH")) or "").lower() or None
+    center_url = _clean(env.get("AGENTFLOW_NODE_CENTER_URL"))
+    # The token is a shared secret compared byte-for-byte on the center; do not
+    # mutate its value, only treat whitespace-only as unset.
+    token = env.get("AGENTFLOW_NODE_TOKEN")
+    if token is not None and token.strip() == "":
+        token = None
+    if not token:
+        raise NodeConfigError(
+            f"AGENTFLOW_NODE_TOKEN is required in role={role!r} — the center "
+            "accepts ingest with it, an edge sends it — but is unset or empty."
+        )
+
+    if role == "center":
+        # The center does not emit, so AGENTFLOW_NODE_CENTER_URL is ignored.
+        if branch is None:
+            branch = CENTER_BRANCH
+        if branch != CENTER_BRANCH:
+            raise NodeConfigError(
+                f"center node AGENTFLOW_NODE_BRANCH must be {CENTER_BRANCH!r}, got {branch!r}."
+            )
+        return NodeConfig(role="center", branch=branch, center_url=None, token=token)
+
+    # role == "edge"
+    if branch not in EDGE_BRANCHES:
+        raise NodeConfigError(
+            "edge node AGENTFLOW_NODE_BRANCH must be one of "
+            f"{sorted(EDGE_BRANCHES)}, got {branch!r}."
+        )
+    if not center_url:
+        raise NodeConfigError(
+            "AGENTFLOW_NODE_CENTER_URL is required in role='edge' (where to emit "
+            "events) but is unset or empty."
+        )
+    return NodeConfig(role="edge", branch=branch, center_url=center_url, token=token)
diff --git a/tests/integration/test_node_topology.py b/tests/integration/test_node_topology.py
new file mode 100644
index 0000000..8fda999
--- /dev/null
+++ b/tests/integration/test_node_topology.py
@@ -0,0 +1,65 @@
+"""Three-node demo topology — full-app node invariants (ADR 0012 / build §13).
+
+N1/N2 here; later steps extend this file with N3-N5/N8/N12 as the ingest
+endpoint, emitter, branch seed, and cross-branch view land. Every case boots
+the real ``app`` via ``TestClient`` (in-memory DuckDB, no Docker).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.serving.api.main import app
+
+pytestmark = pytest.mark.integration
+
+
+def _boot(monkeypatch: pytest.MonkeyPatch, **env: str) -> Iterator[TestClient]:
+    for key, value in env.items():
+        monkeypatch.setenv(key, value)
+    with TestClient(app) as client:
+        yield client
+
+
+@pytest.fixture
+def standalone_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
+    # No AGENTFLOW_NODE_ROLE — today's single-node demo.
+    monkeypatch.delenv("AGENTFLOW_NODE_ROLE", raising=False)
+    monkeypatch.setenv("AGENTFLOW_DEMO_MODE", "true")
+    monkeypatch.setenv("AGENTFLOW_SEED_ON_BOOT", "true")
+    yield from _boot(monkeypatch)
+
+
+def test_standalone_resolves_standalone_role(standalone_client: TestClient) -> None:
+    # N1: unset role == standalone.
+    assert standalone_client.app.state.node_role == "standalone"
+    assert standalone_client.app.state.node_branch is None
+    assert standalone_client.app.state.node_config.is_standalone
+
+
+def test_standalone_does_not_register_ingest_route(standalone_client: TestClient) -> None:
+    # N2: the ingest route is mounted only in center role, so it is absent from
+    # a standalone node's routing table entirely (step 2 adds the center mount
+    # and the full 404/200-by-role HTTP matrix).
+    paths = {getattr(route, "path", None) for route in standalone_client.app.routes}
+    assert "/v1/node/events" not in paths
+
+
+def test_standalone_has_no_emitter_task(standalone_client: TestClient) -> None:
+    # N1: no emitter task in standalone role.
+    assert getattr(standalone_client.app.state, "node_emitter_task", None) is None
+
+
+def test_standalone_health_still_serves(standalone_client: TestClient) -> None:
+    # N1: the existing surface is byte-identical — health still 200.
+    assert standalone_client.get("/v1/health").status_code == 200
+
+
+def test_standalone_ingest_absent_from_public_schema(standalone_client: TestClient) -> None:
+    # The node federation endpoint is internal (token-authed, center-only); it
+    # must never appear in the public agent-facing OpenAPI catalog.
+    schema = standalone_client.get("/openapi.json").json()
+    assert "/v1/node/events" not in schema.get("paths", {})
diff --git a/tests/unit/test_node_config.py b/tests/unit/test_node_config.py
new file mode 100644
index 0000000..efc5f8d
--- /dev/null
+++ b/tests/unit/test_node_config.py
@@ -0,0 +1,197 @@
+"""Node config resolution + fail-fast validation (ADR 0012 / build contract §2).
+
+Pure-logic half of the node invariants: role/branch/center-url/token
+resolution and the boot-time guards. The full-app N1/N2 mounting behaviour is
+in ``tests/integration/test_node_topology.py``.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.serving.node import (
+    CENTER_BRANCH,
+    EDGE_BRANCHES,
+    NodeConfig,
+    NodeConfigError,
+    resolve_node_config,
+)
+
+_NODE_TOKEN = "test-node-token-value"  # noqa: S105 — test fixture, not a real secret
+
+
+def test_standalone_when_role_unset() -> None:
+    cfg = resolve_node_config({})
+    assert cfg.role == "standalone"
+    assert cfg.is_standalone
+    assert cfg.branch is None
+    assert cfg.center_url is None
+    assert cfg.token is None
+
+
+def test_standalone_when_role_blank() -> None:
+    assert resolve_node_config({"AGENTFLOW_NODE_ROLE": "   "}).role == "standalone"
+
+
+def test_standalone_ignores_stray_node_vars() -> None:
+    # Strict superset (N1): a standalone node never validates branch/url/token,
+    # so today's demo Space keeps booting even if unrelated env is present.
+    cfg = resolve_node_config(
+        {
+            "AGENTFLOW_NODE_BRANCH": "spb",
+            "AGENTFLOW_NODE_CENTER_URL": "https://example.test",
+        }
+    )
+    assert cfg.role == "standalone"
+    assert cfg.branch is None
+
+
+def test_role_is_case_insensitive() -> None:
+    cfg = resolve_node_config({"AGENTFLOW_NODE_ROLE": "CENTER", "AGENTFLOW_NODE_TOKEN": "t"})
+    assert cfg.role == "center"
+
+
+def test_invalid_role_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="AGENTFLOW_NODE_ROLE"):
+        resolve_node_config({"AGENTFLOW_NODE_ROLE": "hub", "AGENTFLOW_NODE_TOKEN": "t"})
+
+
+def test_center_defaults_branch_to_msk() -> None:
+    cfg = resolve_node_config(
+        {"AGENTFLOW_NODE_ROLE": "center", "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN}
+    )
+    assert cfg.is_center
+    assert cfg.branch == CENTER_BRANCH
+    assert cfg.center_url is None  # the center does not emit
+    assert cfg.token == _NODE_TOKEN
+
+
+def test_center_explicit_msk_ok() -> None:
+    cfg = resolve_node_config(
+        {
+            "AGENTFLOW_NODE_ROLE": "center",
+            "AGENTFLOW_NODE_BRANCH": "msk",
+            "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+        }
+    )
+    assert cfg.branch == "msk"
+
+
+def test_center_wrong_branch_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="center node"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "center",
+                "AGENTFLOW_NODE_BRANCH": "spb",
+                "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            }
+        )
+
+
+def test_center_without_token_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="AGENTFLOW_NODE_TOKEN"):
+        resolve_node_config({"AGENTFLOW_NODE_ROLE": "center"})
+
+
+def test_center_ignores_center_url() -> None:
+    cfg = resolve_node_config(
+        {
+            "AGENTFLOW_NODE_ROLE": "center",
+            "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            "AGENTFLOW_NODE_CENTER_URL": "https://ignored.test",
+        }
+    )
+    assert cfg.center_url is None
+
+
+@pytest.mark.parametrize("branch", sorted(EDGE_BRANCHES))
+def test_edge_valid(branch: str) -> None:
+    cfg = resolve_node_config(
+        {
+            "AGENTFLOW_NODE_ROLE": "edge",
+            "AGENTFLOW_NODE_BRANCH": branch,
+            "AGENTFLOW_NODE_CENTER_URL": "https://liovina-agentflow-center.hf.space",
+            "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+        }
+    )
+    assert cfg.is_edge
+    assert cfg.branch == branch
+    assert cfg.center_url == "https://liovina-agentflow-center.hf.space"
+    assert cfg.token == _NODE_TOKEN
+
+
+def test_edge_without_center_url_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="AGENTFLOW_NODE_CENTER_URL"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "edge",
+                "AGENTFLOW_NODE_BRANCH": "spb",
+                "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            }
+        )
+
+
+def test_edge_without_token_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="AGENTFLOW_NODE_TOKEN"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "edge",
+                "AGENTFLOW_NODE_BRANCH": "spb",
+                "AGENTFLOW_NODE_CENTER_URL": "https://example.test",
+            }
+        )
+
+
+def test_edge_rejects_center_branch() -> None:
+    # msk is the center's branch; an edge must be a regional warehouse.
+    with pytest.raises(NodeConfigError, match="edge node"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "edge",
+                "AGENTFLOW_NODE_BRANCH": "msk",
+                "AGENTFLOW_NODE_CENTER_URL": "https://example.test",
+                "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            }
+        )
+
+
+def test_edge_rejects_unknown_branch() -> None:
+    with pytest.raises(NodeConfigError, match="edge node"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "edge",
+                "AGENTFLOW_NODE_BRANCH": "tokyo",
+                "AGENTFLOW_NODE_CENTER_URL": "https://example.test",
+                "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            }
+        )
+
+
+def test_edge_missing_branch_fails_fast() -> None:
+    with pytest.raises(NodeConfigError, match="edge node"):
+        resolve_node_config(
+            {
+                "AGENTFLOW_NODE_ROLE": "edge",
+                "AGENTFLOW_NODE_CENTER_URL": "https://example.test",
+                "AGENTFLOW_NODE_TOKEN": _NODE_TOKEN,
+            }
+        )
+
+
+def test_whitespace_only_token_is_unset() -> None:
+    with pytest.raises(NodeConfigError, match="AGENTFLOW_NODE_TOKEN"):
+        resolve_node_config({"AGENTFLOW_NODE_ROLE": "center", "AGENTFLOW_NODE_TOKEN": "   "})
+
+
+def test_token_value_preserved_verbatim() -> None:
+    # The token is compared byte-for-byte on the center; internal whitespace and
+    # surrounding structure must survive resolution untouched.
+    token = "af-node- spaced -secret"  # noqa: S105 — test fixture, not a real secret
+    cfg = resolve_node_config({"AGENTFLOW_NODE_ROLE": "center", "AGENTFLOW_NODE_TOKEN": token})
+    assert cfg.token == token
+
+
+def test_config_is_frozen() -> None:
+    cfg = NodeConfig(role="standalone")
+    with pytest.raises((AttributeError, TypeError)):
+        cfg.role = "center"  # type: ignore[misc]

From 48ffc55fd510d22541eb0d79ff471d8c73995c44 Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 11:52:38 +0300
Subject: [PATCH 2/6] feat(node): center ingest endpoint POST /v1/node/events
 (F2 step 2)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Center-only node federation ingest (ADR 0012 §3). An edge pushes a batch of
the same canonical events; the center applies each via
local_pipeline._process_event — no new serving logic — and tags the
originating branch on the pipeline_events journal (new branch column, NULL
for in-process events).

Auth ladder: role gate (404 off-center, even with a valid token) → bearer
constant-time compare vs AGENTFLOW_NODE_TOKEN (401 missing / 403 wrong) →
body shape (422, unknown origin_branch rejected by Literal). Demo-guard
allow-list + auth-middleware exempt let the token call through while the
public demo-key still fails; endpoint hidden from public OpenAPI. Idempotent
by event_id (re-POST does not double-count). N3/N4/N5/N12 tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/processing/local_pipeline.py        |  50 ++++--
 src/serving/api/auth/middleware.py      |   3 +
 src/serving/api/main.py                 |  10 ++
 src/serving/backends/duckdb_backend.py  |   3 +
 src/serving/node/ingest.py              | 154 ++++++++++++++++++
 tests/integration/test_node_topology.py | 201 +++++++++++++++++++++---
 6 files changed, 384 insertions(+), 37 deletions(-)
 create mode 100644 src/serving/node/ingest.py

diff --git a/src/processing/local_pipeline.py b/src/processing/local_pipeline.py
index 4fa630a..1f34845 100644
--- a/src/processing/local_pipeline.py
+++ b/src/processing/local_pipeline.py
@@ -100,6 +100,9 @@ def _ensure_tables(conn: duckdb.DuckDBPyConnection) -> None:
         "ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS tenant_id VARCHAR DEFAULT 'default'"
     )
     conn.execute("ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS entity_id VARCHAR")
+    # ADR 0012 N4: originating branch of a federated node event (NULL for
+    # in-process events).
+    conn.execute("ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS branch VARCHAR")
 
 
 def _event_tenant(event: dict) -> str:
@@ -109,6 +112,18 @@ def _event_tenant(event: dict) -> str:
     return str(tenant) if tenant else "default"
 
 
+def _event_branch(event: dict) -> str | None:
+    """Originating branch of a federated event (ADR 0012 N4).
+
+    The center's node-ingest endpoint stamps ``source_metadata.branch`` with the
+    edge's ``origin_branch`` before applying, so the ``pipeline_events`` journal
+    carries branch attribution. In-process events (standalone/edge local
+    generator) leave it ``NULL`` — never synthesized."""
+    source_metadata = event.get("source_metadata", {})
+    branch = source_metadata.get("branch") if isinstance(source_metadata, dict) else None
+    return str(branch) if branch else None
+
+
 # ops-surfaces-spec.md §1.3: the entity_id axis of the pipeline_events journal
 # is what Order 360 (and lineage) key off of. NULL when the id isn't
 # derivable from the payload — never synthesized.
@@ -139,6 +154,7 @@ def _process_event(
     event_id = event.get("event_id", "unknown")
     tenant_id = _event_tenant(event)
     entity_id = _derive_entity_id(event, event_type)
+    branch = _event_branch(event)
 
     conn.execute("BEGIN")
     try:
@@ -148,11 +164,12 @@ def _process_event(
             conn.execute(
                 """
                 INSERT INTO pipeline_events (
-                    event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at
+                    event_id, topic, tenant_id, entity_id, event_type, latency_ms,
+                    processed_at, branch
                 )
-                VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?)
+                VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?, ?)
                 """,
-                [event_id, tenant_id, entity_id, event_type, datetime.now(UTC)],
+                [event_id, tenant_id, entity_id, event_type, datetime.now(UTC), branch],
             )
             if iceberg_sink is not None:
                 iceberg_sink.write_batch(
@@ -191,11 +208,12 @@ def _process_event(
             conn.execute(
                 """
                 INSERT INTO pipeline_events (
-                    event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at
+                    event_id, topic, tenant_id, entity_id, event_type, latency_ms,
+                    processed_at, branch
                 )
-                VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?)
+                VALUES (?, 'events.deadletter', ?, ?, ?, 0, ?, ?)
                 """,
-                [event_id, tenant_id, entity_id, event_type, datetime.now(UTC)],
+                [event_id, tenant_id, entity_id, event_type, datetime.now(UTC), branch],
             )
             if iceberg_sink is not None:
                 iceberg_sink.write_batch(
@@ -227,7 +245,7 @@ def _process_event(
         if event_type.startswith("order."):
             event = enrich_order(event)
             _upsert_order(conn, event)
-            _record_order_status(conn, event, event_id, tenant_id)
+            _record_order_status(conn, event, event_id, tenant_id, branch)
             if iceberg_sink is not None:
                 iceberg_sink.write_batch("orders", [event])
         elif event_type in ("click", "page_view", "add_to_cart"):
@@ -257,11 +275,12 @@ def _process_event(
         conn.execute(
             """
             INSERT INTO pipeline_events (
-                event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at
+                event_id, topic, tenant_id, entity_id, event_type, latency_ms,
+                processed_at, branch
             )
-            VALUES (?, 'events.validated', ?, ?, ?, ?, ?)
+            VALUES (?, 'events.validated', ?, ?, ?, ?, ?, ?)
             """,
-            [event_id, tenant_id, entity_id, event_type, latency_ms, datetime.now(UTC)],
+            [event_id, tenant_id, entity_id, event_type, latency_ms, datetime.now(UTC), branch],
         )
         conn.execute("COMMIT")
         if clickhouse_sink is not None:
@@ -334,7 +353,11 @@ def _upsert_order(conn: duckdb.DuckDBPyConnection, event: dict) -> None:
 
 
 def _record_order_status(
-    conn: duckdb.DuckDBPyConnection, event: dict, event_id: str, tenant_id: str
+    conn: duckdb.DuckDBPyConnection,
+    event: dict,
+    event_id: str,
+    tenant_id: str,
+    branch: str | None = None,
 ) -> None:
     """Stage-entry journal row (ops-surfaces-spec.md §1.2) — the stage clock
     for Order 360 / stuck-orders. ``topic='orders.status'`` is deliberately
@@ -343,9 +366,9 @@ def _record_order_status(
     conn.execute(
         """
         INSERT INTO pipeline_events (
-            event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at
+            event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at, branch
         )
-        VALUES (?, 'orders.status', ?, ?, ?, NULL, ?)
+        VALUES (?, 'orders.status', ?, ?, ?, NULL, ?, ?)
         """,
         [
             f"{event_id}-status",
@@ -353,6 +376,7 @@ def _record_order_status(
             str(event["order_id"]),
             f"order.status.{event['status']}",
             datetime.now(UTC),
+            branch,
         ],
     )
 
diff --git a/src/serving/api/auth/middleware.py b/src/serving/api/auth/middleware.py
index a658584..1bccb4a 100644
--- a/src/serving/api/auth/middleware.py
+++ b/src/serving/api/auth/middleware.py
@@ -202,6 +202,9 @@ def _is_exempt_path(path: str) -> bool:
         in {
             "/health",
             "/v1/health",
+            # Node federation ingest (ADR 0012) authenticates with its own
+            # bearer node-token, not an X-API-Key; the endpoint does the check.
+            "/v1/node/events",
         }
     )
 
diff --git a/src/serving/api/main.py b/src/serving/api/main.py
index 3480352..2d8520c 100644
--- a/src/serving/api/main.py
+++ b/src/serving/api/main.py
@@ -57,6 +57,7 @@
 from src.serving.control_plane import control_plane_store_kind, get_control_plane_store
 from src.serving.db_pool import DuckDBPool
 from src.serving.node import resolve_node_config
+from src.serving.node.ingest import router as node_ingest_router
 from src.serving.semantic_layer.catalog import DataCatalog
 from src.serving.semantic_layer.query_engine import QueryEngine
 from src.serving.semantic_layer.search_index import SearchIndex
@@ -299,6 +300,11 @@ async def demo_mode_guard(request: Request, call_next: RequestResponseEndpoint)
         if request.method in {"POST", "PUT", "PATCH", "DELETE"} and path not in {
             "/v1/query",
             "/v1/query/explain",
+            # Node federation ingest (ADR 0012): allow-listed past the demo
+            # read-only guard so the token-authenticated edge->center POST is
+            # not blocked; the endpoint's own bearer check still rejects the
+            # public demo-key caller (N3).
+            "/v1/node/events",
         }:
             return JSONResponse(
                 status_code=403,
@@ -347,6 +353,10 @@ async def demo_mode_guard(request: Request, call_next: RequestResponseEndpoint)
 app.include_router(slo_router)
 app.include_router(stream_router)
 app.include_router(webhook_router)
+# Three-node topology (ADR 0012): the node-ingest router is mounted on every
+# node but is a no-op (404) off the center and hidden from the public OpenAPI
+# (include_in_schema=False). It carries its own bearer-token auth.
+app.include_router(node_ingest_router)
 
 
 @app.get("/v1/changelog", response_model=None)
diff --git a/src/serving/backends/duckdb_backend.py b/src/serving/backends/duckdb_backend.py
index 182442c..3980e94 100644
--- a/src/serving/backends/duckdb_backend.py
+++ b/src/serving/backends/duckdb_backend.py
@@ -185,6 +185,9 @@ def initialize_demo_data(self) -> None:
         self._conn.execute(
             "ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS latency_ms INTEGER"
         )
+        # ADR 0012 N4: federated node events carry their originating branch on
+        # the journal; NULL for the standalone demo's in-process events.
+        self._conn.execute("ALTER TABLE pipeline_events ADD COLUMN IF NOT EXISTS branch VARCHAR")
 
         row = self._conn.execute("SELECT COUNT(*) FROM orders_v2").fetchone()
         count = row[0] if row else 0
diff --git a/src/serving/node/ingest.py b/src/serving/node/ingest.py
new file mode 100644
index 0000000..1312ceb
--- /dev/null
+++ b/src/serving/node/ingest.py
@@ -0,0 +1,154 @@
+"""Center node-ingest endpoint — ``POST /v1/node/events`` (ADR 0012 §3 / build §4).
+
+An edge branch pushes a batch of the **same canonical events** the in-process
+pipeline already understands; the center applies each through
+:func:`local_pipeline._process_event` — no new serving logic — tagging the
+originating branch on the ``pipeline_events`` journal so the cross-branch view
+(step 5), Order 360, and freshness move.
+
+The endpoint is mounted on every node but is a no-op (``404``) off the center,
+carries its own bearer-token auth (distinct from the public ``demo-key``), and
+is hidden from the public OpenAPI catalog (``include_in_schema=False``) because
+it is internal node-to-node federation, not an agent-facing surface.
+"""
+
+from __future__ import annotations
+
+import json
+import secrets
+import threading
+from typing import Literal
+
+import duckdb
+import structlog
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field, ValidationError
+from starlette.concurrency import run_in_threadpool
+
+from src.processing.local_pipeline import _process_event
+
+logger = structlog.get_logger()
+
+router = APIRouter(prefix="/v1/node", tags=["node"])
+
+# Bounded batch (build contract §4): a demo edge emits a handful of events per
+# request, never a load-test flood.
+MAX_EVENTS_PER_BATCH = 500
+
+# Ingest writes go through the shared serving write-connection; serialize them
+# so two concurrent POSTs never interleave BEGIN/COMMIT on it. One center =
+# one process = one lock.
+_INGEST_LOCK = threading.Lock()
+
+
+class NodeEventBatch(BaseModel):
+    """One edge->center push. ``origin_branch`` is constrained to the live edge
+    branches, so an unknown/mismatched branch is rejected as ``422`` (N12)."""
+
+    origin_branch: Literal["spb", "ekb"]
+    events: list[dict] = Field(default_factory=list, max_length=MAX_EVENTS_PER_BATCH)
+
+
+def _extract_bearer(request: Request) -> str | None:
+    header = request.headers.get("authorization")
+    if not header:
+        return None
+    scheme, _, value = header.partition(" ")
+    if scheme.lower() != "bearer" or not value.strip():
+        return None
+    return value.strip()
+
+
+def _existing_event_ids(conn: duckdb.DuckDBPyConnection, event_ids: list[str]) -> set[str]:
+    """Batch ids already present in the ingest journal — the idempotency filter
+    (N5). A static query over the two ingest topics (the derived
+    ``orders.status`` row uses a distinct id/topic); intersected with the batch
+    in Python so the SQL carries no interpolated values."""
+    if not event_ids:
+        return set()
+    rows = conn.execute(
+        "SELECT DISTINCT event_id FROM pipeline_events "
+        "WHERE topic IN ('events.validated', 'events.deadletter')"
+    ).fetchall()
+    return {str(row[0]) for row in rows} & set(event_ids)
+
+
+@router.post("/events", include_in_schema=False, response_model=None)
+async def ingest_node_events(request: Request) -> JSONResponse:
+    config = request.app.state.node_config
+
+    # N2/N12: a usable ingest exists only on the center — every other role
+    # (edge, standalone) is 404, even with a valid token.
+    if not config.is_center:
+        return JSONResponse(status_code=404, content={"detail": "Not Found"})
+
+    # Bearer auth against the node token, constant-time, distinct from the
+    # public demo-key path (N3/N10). Role gate above runs first so a non-center
+    # node never even reveals the auth ladder.
+    bearer = _extract_bearer(request)
+    if bearer is None:
+        return JSONResponse(
+            status_code=401, content={"detail": "Missing or malformed bearer token."}
+        )
+    expected = config.token or ""
+    if not expected or not secrets.compare_digest(bearer, expected):
+        return JSONResponse(status_code=403, content={"detail": "Invalid node token."})
+
+    # Body shape (422); an unknown origin_branch is rejected here by the Literal.
+    try:
+        raw = await request.json()
+        batch = NodeEventBatch.model_validate(raw)
+    except (json.JSONDecodeError, ValidationError, ValueError, TypeError) as exc:
+        return JSONResponse(status_code=422, content={"detail": f"Invalid batch: {exc}"})
+
+    origin = batch.origin_branch
+    events = batch.events
+    conn = request.app.state.query_engine._conn
+
+    def _apply() -> tuple[int, int, int]:
+        with _INGEST_LOCK:
+            ids = [str(e["event_id"]) for e in events if isinstance(e, dict) and e.get("event_id")]
+            seen = _existing_event_ids(conn, ids)
+            applied = dead = duplicates = 0
+            for event in events:
+                event_id = event.get("event_id") if isinstance(event, dict) else None
+                if event_id is not None and str(event_id) in seen:
+                    # Idempotency (N5): re-POSTing the same batch never
+                    # re-applies or re-journals; count it, do not double-count.
+                    duplicates += 1
+                    continue
+                # Tag origin so the journal/lineage carry the branch (N4).
+                metadata = event.setdefault("source_metadata", {})
+                if isinstance(metadata, dict):
+                    metadata["branch"] = origin
+                # No new serving logic — the center reuses the exact event->metric
+                # path the in-process pipeline uses. clickhouse_sink is None: the
+                # HF three-node demo serves from DuckDB.
+                success, _reason = _process_event(conn, event, clickhouse_sink=None)
+                if success:
+                    applied += 1
+                    if event_id is not None:
+                        seen.add(str(event_id))
+                else:
+                    dead += 1
+            return applied, dead, duplicates
+
+    applied, dead_lettered, duplicates = await run_in_threadpool(_apply)
+    logger.info(
+        "node_events_ingested",
+        origin_branch=origin,
+        accepted=len(events),
+        applied=applied,
+        dead_lettered=dead_lettered,
+        duplicates=duplicates,
+    )
+    return JSONResponse(
+        status_code=200,
+        content={
+            "accepted": len(events),
+            "applied": applied,
+            "dead_lettered": dead_lettered,
+            "duplicates": duplicates,
+        },
+    )
diff --git a/tests/integration/test_node_topology.py b/tests/integration/test_node_topology.py
index 8fda999..5749f3f 100644
--- a/tests/integration/test_node_topology.py
+++ b/tests/integration/test_node_topology.py
@@ -1,65 +1,218 @@
 """Three-node demo topology — full-app node invariants (ADR 0012 / build §13).
 
-N1/N2 here; later steps extend this file with N3-N5/N8/N12 as the ingest
-endpoint, emitter, branch seed, and cross-branch view land. Every case boots
-the real ``app`` via ``TestClient`` (in-memory DuckDB, no Docker).
+Every case boots the real ``app`` via ``TestClient`` (in-memory DuckDB, no
+Docker). Covered here: N1 (standalone unchanged), N2 (mounted iff center),
+N3 (bearer auth vs demo-key), N4 (apply + branch tag), N5 (idempotency),
+N12 (role/branch guard). N6/N8/N11 land with the seed + cross-branch view.
 """
 
 from __future__ import annotations
 
+import json
 from collections.abc import Iterator
 
 import pytest
 from fastapi.testclient import TestClient
 
+from src.ingestion.producers.event_producer import generate_order
 from src.serving.api.main import app
 
 pytestmark = pytest.mark.integration
 
+_TOKEN = "test-center-node-token"  # noqa: S105 — test fixture, not a real secret
+_AUTH = {"Authorization": f"Bearer {_TOKEN}"}
 
-def _boot(monkeypatch: pytest.MonkeyPatch, **env: str) -> Iterator[TestClient]:
-    for key, value in env.items():
-        monkeypatch.setenv(key, value)
-    with TestClient(app) as client:
-        yield client
+
+def _order_event() -> dict:
+    """A real canonical order event (producer-shaped, valid uuid event_id) — the
+    exact dict the edge emitter forwards. Ids are the producer's own so schema
+    validation passes; tests read them back off the returned dict."""
+    _topic, model = generate_order()
+    return json.loads(model.model_dump_json())
 
 
 @pytest.fixture
 def standalone_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
-    # No AGENTFLOW_NODE_ROLE — today's single-node demo.
     monkeypatch.delenv("AGENTFLOW_NODE_ROLE", raising=False)
     monkeypatch.setenv("AGENTFLOW_DEMO_MODE", "true")
     monkeypatch.setenv("AGENTFLOW_SEED_ON_BOOT", "true")
-    yield from _boot(monkeypatch)
+    with TestClient(app) as client:
+        yield client
+
+
+@pytest.fixture
+def center_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
+    monkeypatch.setenv("AGENTFLOW_NODE_ROLE", "center")
+    monkeypatch.setenv("AGENTFLOW_NODE_BRANCH", "msk")
+    monkeypatch.setenv("AGENTFLOW_NODE_TOKEN", _TOKEN)
+    monkeypatch.setenv("AGENTFLOW_DEMO_MODE", "true")
+    monkeypatch.setenv("AGENTFLOW_SEED_ON_BOOT", "true")
+    with TestClient(app) as client:
+        yield client
+
+
+@pytest.fixture
+def edge_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
+    monkeypatch.setenv("AGENTFLOW_NODE_ROLE", "edge")
+    monkeypatch.setenv("AGENTFLOW_NODE_BRANCH", "spb")
+    monkeypatch.setenv("AGENTFLOW_NODE_CENTER_URL", "https://center.invalid")
+    monkeypatch.setenv("AGENTFLOW_NODE_TOKEN", _TOKEN)
+    monkeypatch.setenv("AGENTFLOW_NODE_EMITTER_ENABLED", "false")
+    monkeypatch.setenv("AGENTFLOW_DEMO_MODE", "true")
+    monkeypatch.setenv("AGENTFLOW_SEED_ON_BOOT", "true")
+    with TestClient(app) as client:
+        yield client
+
+
+# --- N1: standalone is byte-identical -------------------------------------
 
 
 def test_standalone_resolves_standalone_role(standalone_client: TestClient) -> None:
-    # N1: unset role == standalone.
     assert standalone_client.app.state.node_role == "standalone"
     assert standalone_client.app.state.node_branch is None
     assert standalone_client.app.state.node_config.is_standalone
 
 
-def test_standalone_does_not_register_ingest_route(standalone_client: TestClient) -> None:
-    # N2: the ingest route is mounted only in center role, so it is absent from
-    # a standalone node's routing table entirely (step 2 adds the center mount
-    # and the full 404/200-by-role HTTP matrix).
-    paths = {getattr(route, "path", None) for route in standalone_client.app.routes}
-    assert "/v1/node/events" not in paths
-
-
 def test_standalone_has_no_emitter_task(standalone_client: TestClient) -> None:
-    # N1: no emitter task in standalone role.
     assert getattr(standalone_client.app.state, "node_emitter_task", None) is None
 
 
 def test_standalone_health_still_serves(standalone_client: TestClient) -> None:
-    # N1: the existing surface is byte-identical — health still 200.
     assert standalone_client.get("/v1/health").status_code == 200
 
 
-def test_standalone_ingest_absent_from_public_schema(standalone_client: TestClient) -> None:
-    # The node federation endpoint is internal (token-authed, center-only); it
-    # must never appear in the public agent-facing OpenAPI catalog.
+def test_ingest_absent_from_public_schema(standalone_client: TestClient) -> None:
+    # Internal node-to-node endpoint — never in the public agent catalog.
     schema = standalone_client.get("/openapi.json").json()
     assert "/v1/node/events" not in schema.get("paths", {})
+
+
+# --- N2: mounted iff role == center ---------------------------------------
+
+
+def test_standalone_ingest_is_404(standalone_client: TestClient) -> None:
+    resp = standalone_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": []}, headers=_AUTH
+    )
+    assert resp.status_code == 404
+
+
+def test_edge_ingest_is_404_even_with_token(edge_client: TestClient) -> None:
+    # N12: a non-center node refuses ingest even with a valid token.
+    resp = edge_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": []}, headers=_AUTH
+    )
+    assert resp.status_code == 404
+
+
+def test_center_ingest_accepts_empty_batch(center_client: TestClient) -> None:
+    resp = center_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": []}, headers=_AUTH
+    )
+    assert resp.status_code == 200
+    assert resp.json() == {"accepted": 0, "applied": 0, "dead_lettered": 0, "duplicates": 0}
+
+
+# --- N3 / N10: bearer auth, not the demo-key ------------------------------
+
+
+def test_center_ingest_missing_bearer_is_401(center_client: TestClient) -> None:
+    resp = center_client.post("/v1/node/events", json={"origin_branch": "spb", "events": []})
+    assert resp.status_code == 401
+
+
+def test_center_ingest_public_demo_key_is_rejected(center_client: TestClient) -> None:
+    # The public demo-key (X-API-Key) is not the node token — no bearer → 401.
+    resp = center_client.post(
+        "/v1/node/events",
+        json={"origin_branch": "spb", "events": []},
+        headers={"X-API-Key": "demo-key"},
+    )
+    assert resp.status_code == 401
+
+
+def test_center_ingest_wrong_token_is_403(center_client: TestClient) -> None:
+    resp = center_client.post(
+        "/v1/node/events",
+        json={"origin_branch": "spb", "events": []},
+        headers={"Authorization": "Bearer not-the-token"},
+    )
+    assert resp.status_code == 403
+
+
+# --- N4: apply via _process_event, branch-tag the journal, move the metric -
+
+
+def test_center_applies_event_and_tags_branch(center_client: TestClient) -> None:
+    event = _order_event()
+    resp = center_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": [event]}, headers=_AUTH
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["accepted"] == 1
+    assert body["applied"] == 1
+    assert body["dead_lettered"] == 0
+
+    conn = center_client.app.state.query_engine._conn
+    # Journal row carries branch=spb (N4).
+    branches = conn.execute(
+        "SELECT DISTINCT branch FROM pipeline_events "
+        "WHERE event_id = ? AND topic = 'events.validated'",
+        [event["event_id"]],
+    ).fetchall()
+    assert branches == [("spb",)]
+    # The order landed on the center's serving surface (metric moves).
+    order_count = conn.execute(
+        "SELECT COUNT(*) FROM orders_v2 WHERE order_id = ?", [event["order_id"]]
+    ).fetchone()
+    assert order_count is not None
+    assert order_count[0] == 1
+
+
+# --- N5: idempotency — re-POST the same batch does not double-count --------
+
+
+def test_center_ingest_is_idempotent(center_client: TestClient) -> None:
+    event = _order_event()
+    batch = {"origin_branch": "spb", "events": [event]}
+
+    first = center_client.post("/v1/node/events", json=batch, headers=_AUTH)
+    assert first.status_code == 200
+    assert first.json()["applied"] == 1
+
+    second = center_client.post("/v1/node/events", json=batch, headers=_AUTH)
+    assert second.status_code == 200
+    body = second.json()
+    assert body["applied"] == 0
+    assert body["duplicates"] == 1
+
+    conn = center_client.app.state.query_engine._conn
+    validated = conn.execute(
+        "SELECT COUNT(*) FROM pipeline_events WHERE event_id = ? AND topic = 'events.validated'",
+        [event["event_id"]],
+    ).fetchone()
+    assert validated is not None
+    assert validated[0] == 1
+
+
+# --- N12: role/branch guard -----------------------------------------------
+
+
+def test_center_rejects_unknown_origin_branch(center_client: TestClient) -> None:
+    resp = center_client.post(
+        "/v1/node/events",
+        json={"origin_branch": "tokyo", "events": []},
+        headers=_AUTH,
+    )
+    assert resp.status_code == 422
+
+
+def test_center_rejects_oversized_batch(center_client: TestClient) -> None:
+    # The size bound is a body-shape guard (422) checked before any apply, so
+    # the event contents are irrelevant here.
+    events = [{} for _ in range(501)]
+    resp = center_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": events}, headers=_AUTH
+    )
+    assert resp.status_code == 422

From 3aa5ef13ac5cdfd686ddb9a99e6c182f4e22a6b9 Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 11:59:38 +0300
Subject: [PATCH 3/6] =?UTF-8?q?feat(node):=20edge=20emitter=20=E2=80=94=20?=
 =?UTF-8?q?local-apply=20+=20forward=20to=20center=20(F2=20step=203)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Edge-role background task (ADR 0012 §6). A slow generator produces the same
canonical events the in-process pipeline makes; each is stamped with the
edge's branch, applied to the edge's own read surface, and forwarded
unchanged to the center's POST /v1/node/events with the node token (N7).

Cold-center tolerance (N9): bounded retries with backoff, drop + log on
give-up, never raised into the loop — the edge stays up when the hub is
asleep, which is what makes the push-on-activity sleep choreography
self-healing. Started in lifespan only for edge role (off in center/
standalone; AGENTFLOW_NODE_EMITTER_ENABLED=false disables for tests).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/serving/api/main.py                 |  32 ++++++
 src/serving/node/emitter.py             | 138 ++++++++++++++++++++++++
 tests/integration/test_node_topology.py |  31 ++++++
 tests/unit/test_node_emitter.py         | 118 ++++++++++++++++++++
 4 files changed, 319 insertions(+)
 create mode 100644 src/serving/node/emitter.py
 create mode 100644 tests/unit/test_node_emitter.py

diff --git a/src/serving/api/main.py b/src/serving/api/main.py
index 2d8520c..df5d80d 100644
--- a/src/serving/api/main.py
+++ b/src/serving/api/main.py
@@ -57,6 +57,7 @@
 from src.serving.control_plane import control_plane_store_kind, get_control_plane_store
 from src.serving.db_pool import DuckDBPool
 from src.serving.node import resolve_node_config
+from src.serving.node.emitter import NodeEmitter
 from src.serving.node.ingest import router as node_ingest_router
 from src.serving.semantic_layer.catalog import DataCatalog
 from src.serving.semantic_layer.query_engine import QueryEngine
@@ -80,6 +81,20 @@ def setup_telemetry(
 logger = structlog.get_logger()
 
 
+def _env_float(name: str, fallback: float) -> float:
+    try:
+        return float(os.environ[name])
+    except (KeyError, ValueError):
+        return fallback
+
+
+def _env_int(name: str, fallback: int) -> int:
+    try:
+        return int(os.environ[name])
+    except (KeyError, ValueError):
+        return fallback
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     """Initialize shared resources on startup."""
@@ -249,6 +264,21 @@ async def dispatch_new_events_with_cache_invalidation() -> None:
         )
     app.state.outbox_processor_task = asyncio.create_task(app.state.outbox_processor.run_forever())
 
+    # Edge role (ADR 0012): start the slow generator->forward emitter. Off in
+    # center/standalone; tests disable it with AGENTFLOW_NODE_EMITTER_ENABLED=false.
+    app.state.node_emitter = None
+    app.state.node_emitter_task = None
+    emitter_enabled = os.getenv("AGENTFLOW_NODE_EMITTER_ENABLED", "true").lower() != "false"
+    if app.state.node_config.is_edge and emitter_enabled:
+        app.state.node_emitter = NodeEmitter(
+            config=app.state.node_config,
+            conn=app.state.query_engine._conn,
+            interval_seconds=_env_float("AGENTFLOW_NODE_EMIT_INTERVAL_SECONDS", 3.0),
+            batch_size=_env_int("AGENTFLOW_NODE_EMIT_BATCH_SIZE", 5),
+        )
+        app.state.node_emitter.start()
+        app.state.node_emitter_task = app.state.node_emitter.task
+
     auth_mode = (
         "multi_tenant_api_keys"
         if app.state.auth_manager.has_configured_keys()
@@ -273,6 +303,8 @@ async def dispatch_new_events_with_cache_invalidation() -> None:
         pass
     await app.state.alert_dispatcher.stop()
     await app.state.webhook_dispatcher.stop()
+    if getattr(app.state, "node_emitter", None) is not None:
+        await app.state.node_emitter.stop()
     await app.state.query_cache.close()
     app.state.db_pool.close()
     logger.info("api_shutting_down")
diff --git a/src/serving/node/emitter.py b/src/serving/node/emitter.py
new file mode 100644
index 0000000..51ee988
--- /dev/null
+++ b/src/serving/node/emitter.py
@@ -0,0 +1,138 @@
+"""Edge emitter — pushes an edge branch's live events to the center (ADR 0012 §6).
+
+In edge role a slow background generator produces the same canonical events the
+in-process pipeline already makes; each is applied to the edge's own read
+surface (so the branch Space stays live) and forwarded, unchanged, to the
+center's ``POST /v1/node/events``. This is push-on-activity, which is what makes
+the sleep choreography self-healing: a cold/asleep center is tolerated (bounded
+retries, drop on give-up), never raised into the loop, so the edge page stays up
+even when the hub is down.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from collections.abc import Callable
+
+import duckdb
+import httpx
+import structlog
+
+from src.processing.local_pipeline import _generate_random_event, _process_event
+from src.serving.node.config import NodeConfig
+
+logger = structlog.get_logger()
+
+
+class NodeEmitter:
+    """Edge-role background task. Constructed with injectable dependencies so the
+    forward path and the produce/apply seam are unit-testable without a network
+    or a live loop."""
+
+    def __init__(
+        self,
+        *,
+        config: NodeConfig,
+        conn: duckdb.DuckDBPyConnection,
+        client_factory: Callable[[], httpx.AsyncClient] | None = None,
+        interval_seconds: float = 3.0,
+        batch_size: int = 5,
+        timeout_seconds: float = 5.0,
+        max_retries: int = 3,
+        backoff_base_seconds: float = 0.5,
+    ) -> None:
+        self._config = config
+        self._conn = conn
+        self._interval = interval_seconds
+        self._batch_size = max(1, batch_size)
+        self._timeout = timeout_seconds
+        self._max_retries = max(1, max_retries)
+        self._backoff_base = backoff_base_seconds
+        self._client_factory = client_factory or self._default_client_factory
+        self._task: asyncio.Task[None] | None = None
+        self._stopping = asyncio.Event()
+
+    def _default_client_factory(self) -> httpx.AsyncClient:
+        # The center URL is operator-configured (trusted), not a user-supplied
+        # webhook target, so it deliberately does not go through the SSRF egress
+        # guard the webhook dispatcher uses.
+        return httpx.AsyncClient(timeout=self._timeout)
+
+    @property
+    def endpoint(self) -> str:
+        base = (self._config.center_url or "").rstrip("/")
+        return f"{base}/v1/node/events"
+
+    @property
+    def task(self) -> asyncio.Task[None] | None:
+        return self._task
+
+    def _produce_local(self) -> dict:
+        """Generate one event, stamp it with this edge's branch, apply it to the
+        edge's own read surface, and return the **same** dict that will be
+        forwarded (N7). Blocking DuckDB work — call via a worker thread."""
+        _topic, event = _generate_random_event()
+        metadata = event.setdefault("source_metadata", {})
+        if isinstance(metadata, dict):
+            metadata["branch"] = self._config.branch
+        _process_event(self._conn, event, clickhouse_sink=None)
+        return event
+
+    async def _forward(self, client: httpx.AsyncClient, events: list[dict]) -> bool:
+        """POST a batch to the center. Tolerant of a cold/asleep center (N9):
+        bounded retries with backoff, drop + log on give-up, **never** raise into
+        the loop."""
+        if not events:
+            return True
+        payload = {"origin_branch": self._config.branch, "events": events}
+        headers = {"Authorization": f"Bearer {self._config.token or ''}"}
+        for attempt in range(1, self._max_retries + 1):
+            try:
+                response = await client.post(self.endpoint, json=payload, headers=headers)
+            except (httpx.TimeoutException, httpx.TransportError) as exc:
+                logger.info("node_emit_retry", attempt=attempt, error=str(exc))
+                if attempt < self._max_retries:
+                    await asyncio.sleep(self._backoff_base * attempt)
+                continue
+            if response.status_code == 200:
+                return True
+            # A rejected batch (auth/shape) will not succeed on retry — drop it.
+            logger.info("node_emit_rejected", status=response.status_code)
+            return False
+        logger.info("node_emit_dropped", branch=self._config.branch, count=len(events))
+        return False
+
+    async def _run(self) -> None:
+        buffer: list[dict] = []
+        async with self._client_factory() as client:
+            while not self._stopping.is_set():
+                try:
+                    event = await asyncio.to_thread(self._produce_local)
+                    buffer.append(event)
+                    if len(buffer) >= self._batch_size:
+                        await self._forward(client, buffer)
+                        buffer = []
+                except Exception:  # noqa: BLE001 — the loop must never die
+                    logger.warning("node_emit_loop_error", exc_info=True)
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(self._stopping.wait(), timeout=self._interval)
+
+    def start(self) -> None:
+        if self._task is None:
+            self._stopping.clear()
+            self._task = asyncio.create_task(self._run())
+            logger.info(
+                "node_emitter_started",
+                branch=self._config.branch,
+                center=self.endpoint,
+                interval_seconds=self._interval,
+            )
+
+    async def stop(self) -> None:
+        self._stopping.set()
+        if self._task is not None:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._task
+            self._task = None
diff --git a/tests/integration/test_node_topology.py b/tests/integration/test_node_topology.py
index 5749f3f..e1d5f27 100644
--- a/tests/integration/test_node_topology.py
+++ b/tests/integration/test_node_topology.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 from collections.abc import Iterator
 
@@ -64,6 +65,23 @@ def edge_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
         yield client
 
 
+@pytest.fixture
+def edge_emitting_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
+    # Emitter on, but a long interval + large batch so it applies at most one
+    # event locally and never forwards during the test (no network to the
+    # unreachable center).
+    monkeypatch.setenv("AGENTFLOW_NODE_ROLE", "edge")
+    monkeypatch.setenv("AGENTFLOW_NODE_BRANCH", "ekb")
+    monkeypatch.setenv("AGENTFLOW_NODE_CENTER_URL", "https://center.invalid")
+    monkeypatch.setenv("AGENTFLOW_NODE_TOKEN", _TOKEN)
+    monkeypatch.setenv("AGENTFLOW_NODE_EMIT_INTERVAL_SECONDS", "3600")
+    monkeypatch.setenv("AGENTFLOW_NODE_EMIT_BATCH_SIZE", "1000")
+    monkeypatch.setenv("AGENTFLOW_DEMO_MODE", "true")
+    monkeypatch.setenv("AGENTFLOW_SEED_ON_BOOT", "true")
+    with TestClient(app) as client:
+        yield client
+
+
 # --- N1: standalone is byte-identical -------------------------------------
 
 
@@ -113,6 +131,19 @@ def test_center_ingest_accepts_empty_batch(center_client: TestClient) -> None:
     assert resp.json() == {"accepted": 0, "applied": 0, "dead_lettered": 0, "duplicates": 0}
 
 
+# --- Emitter wiring: started only in edge role ----------------------------
+
+
+def test_edge_boot_starts_emitter(edge_emitting_client: TestClient) -> None:
+    task = edge_emitting_client.app.state.node_emitter_task
+    assert isinstance(task, asyncio.Task)
+    assert edge_emitting_client.app.state.node_emitter is not None
+
+
+def test_center_boot_has_no_emitter(center_client: TestClient) -> None:
+    assert getattr(center_client.app.state, "node_emitter_task", None) is None
+
+
 # --- N3 / N10: bearer auth, not the demo-key ------------------------------
 
 
diff --git a/tests/unit/test_node_emitter.py b/tests/unit/test_node_emitter.py
new file mode 100644
index 0000000..28867d2
--- /dev/null
+++ b/tests/unit/test_node_emitter.py
@@ -0,0 +1,118 @@
+"""Edge emitter — produce/apply seam and cold-center tolerance (ADR 0012 §6).
+
+Unit-level (no app boot, no network): an in-memory DuckDB for the local apply
+and a hand-rolled async client for the forward path. Covers N7 (same dict
+applied locally and forwarded) and N9 (a cold/unreachable center is tolerated,
+never raised).
+"""
+
+from __future__ import annotations
+
+import duckdb
+import httpx
+
+from src.processing.local_pipeline import _ensure_tables
+from src.serving.node.config import NodeConfig
+from src.serving.node.emitter import NodeEmitter
+
+_TOKEN = "edge-node-token"  # noqa: S105 — test fixture, not a real secret
+
+
+def _edge_config() -> NodeConfig:
+    return NodeConfig(role="edge", branch="spb", center_url="https://center.example/", token=_TOKEN)
+
+
+def _emitter(**kwargs) -> tuple[NodeEmitter, duckdb.DuckDBPyConnection]:
+    conn = duckdb.connect(":memory:")
+    _ensure_tables(conn)
+    return NodeEmitter(config=_edge_config(), conn=conn, **kwargs), conn
+
+
+class _CapturingClient:
+    """Minimal async client double capturing the last POST."""
+
+    def __init__(self, status_code: int = 200) -> None:
+        self.status_code = status_code
+        self.calls: list[dict] = []
+
+    async def post(self, url: str, json: dict, headers: dict) -> httpx.Response:
+        self.calls.append({"url": url, "json": json, "headers": headers})
+        return httpx.Response(self.status_code)
+
+
+class _ColdClient:
+    """Async client double that always fails the connection (cold/asleep hub)."""
+
+    def __init__(self) -> None:
+        self.attempts = 0
+
+    async def post(self, url: str, json: dict, headers: dict) -> httpx.Response:
+        self.attempts += 1
+        raise httpx.ConnectError("center is asleep")
+
+
+# --- N7: local apply and forwarded payload are the same canonical dict -----
+
+
+def test_produce_local_tags_branch_and_applies() -> None:
+    emitter, conn = _emitter()
+    event = emitter._produce_local()
+
+    assert event["source_metadata"]["branch"] == "spb"
+    row = conn.execute(
+        "SELECT branch FROM pipeline_events "
+        "WHERE event_id = ? AND topic IN ('events.validated', 'events.deadletter')",
+        [event["event_id"]],
+    ).fetchone()
+    assert row is not None
+    assert row[0] == "spb"
+
+
+async def test_forward_posts_the_same_events_verbatim() -> None:
+    emitter, _conn = _emitter()
+    client = _CapturingClient()
+    event = {"event_id": "evt-x", "event_type": "order.created"}
+
+    ok = await emitter._forward(client, [event])
+
+    assert ok is True
+    assert len(client.calls) == 1
+    call = client.calls[0]
+    assert call["url"] == "https://center.example/v1/node/events"
+    assert call["json"]["origin_branch"] == "spb"
+    assert call["json"]["events"] == [event]  # forwarded unchanged (N7)
+    assert call["headers"]["Authorization"] == f"Bearer {_TOKEN}"
+
+
+# --- N9: cold-center tolerance ---------------------------------------------
+
+
+async def test_forward_tolerates_cold_center() -> None:
+    emitter, _conn = _emitter(max_retries=3, backoff_base_seconds=0.0)
+    client = _ColdClient()
+
+    # Must not raise even though every attempt fails.
+    ok = await emitter._forward(client, [{"event_id": "evt-x"}])
+
+    assert ok is False
+    assert client.attempts == 3  # bounded retries, then drop
+
+
+async def test_forward_drops_rejected_batch_without_retry() -> None:
+    emitter, _conn = _emitter(max_retries=3, backoff_base_seconds=0.0)
+    client = _CapturingClient(status_code=403)
+
+    ok = await emitter._forward(client, [{"event_id": "evt-x"}])
+
+    assert ok is False
+    assert len(client.calls) == 1  # a rejected batch is not retried
+
+
+async def test_forward_empty_batch_is_noop() -> None:
+    emitter, _conn = _emitter()
+    client = _CapturingClient()
+
+    ok = await emitter._forward(client, [])
+
+    assert ok is True
+    assert client.calls == []

From 5466d868c1caa94dbe8e729dc5ca7b69bce8df3c Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 12:03:29 +0300
Subject: [PATCH 4/6] feat(node): branch-scoped baseline seed (F2 step 4)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Boot baseline for the cross-branch view (ADR 0012 §7). The demo entity tables
are a shared catalog with no branch column, so branch attribution lives on the
pipeline_events journal: a deterministic per-branch baseline under a distinct
topic (node.baseline). Center seeds all three branches, an edge seeds only its
own (N6), standalone seeds none — byte-identical to today's demo (N1). Pure
function of the branch, cleared-then-relaid, so a restart yields the same
baseline (N11). The distinct topic keeps the rows inert to every existing
journal scan and separable from the live delta.

N6/N11 unit + integration tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/serving/api/main.py                 |  5 ++
 src/serving/node/seed.py                | 63 +++++++++++++++++++
 tests/integration/test_node_topology.py | 25 ++++++++
 tests/unit/test_node_seed.py            | 82 +++++++++++++++++++++++++
 4 files changed, 175 insertions(+)
 create mode 100644 src/serving/node/seed.py
 create mode 100644 tests/unit/test_node_seed.py

diff --git a/src/serving/api/main.py b/src/serving/api/main.py
index df5d80d..f147393 100644
--- a/src/serving/api/main.py
+++ b/src/serving/api/main.py
@@ -59,6 +59,7 @@
 from src.serving.node import resolve_node_config
 from src.serving.node.emitter import NodeEmitter
 from src.serving.node.ingest import router as node_ingest_router
+from src.serving.node.seed import seed_node_baseline
 from src.serving.semantic_layer.catalog import DataCatalog
 from src.serving.semantic_layer.query_engine import QueryEngine
 from src.serving.semantic_layer.search_index import SearchIndex
@@ -147,6 +148,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         app.state.query_engine._duckdb_backend.initialize_demo_data()
         if app.state.query_engine._backend_name != app.state.query_engine._duckdb_backend.name:
             app.state.query_engine._backend.initialize_demo_data()
+        # Three-node topology (ADR 0012 §7): lay down the per-branch journal
+        # baseline (center = all branches, edge = its own, standalone = none)
+        # so a center-first visitor sees a coherent cross-branch picture.
+        seed_node_baseline(app.state.query_engine._conn, app.state.node_config)
     app.state.search_index = SearchIndex(
         catalog=app.state.catalog,
         query_engine=app.state.query_engine,
diff --git a/src/serving/node/seed.py b/src/serving/node/seed.py
new file mode 100644
index 0000000..02d25a7
--- /dev/null
+++ b/src/serving/node/seed.py
@@ -0,0 +1,63 @@
+"""Branch-scoped boot baseline for the three-node demo (ADR 0012 §7, N6/N11).
+
+The demo entity tables (``orders_v2`` …) are a shared catalog with no branch
+column, so branch attribution lives on the ``pipeline_events`` journal. This
+seed lays down a deterministic per-branch **baseline** under a distinct topic
+(``node.baseline``) so a center-first visitor sees a coherent cross-branch
+picture before any live event arrives (N8 baseline half). It is a pure function
+of the branch, so any node re-seeds to the same baseline on every restart (N11);
+standalone gets none, keeping today's demo byte-identical (N1).
+
+The distinct topic keeps these rows inert to every existing journal scan
+(freshness/webhook/lineage/stage-clock all filter on their own topics), and lets
+the cross-branch view (step 5) tell the seeded baseline apart from the live
+delta.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import duckdb
+
+from src.serving.node.config import NodeConfig
+
+NODE_BASELINE_TOPIC = "node.baseline"
+
+# The center's HQ (msk) carries the largest baseline; the two RU regional
+# warehouses are smaller. Fixed figures → a deterministic baseline (N11).
+BASELINE_ROWS_BY_BRANCH: dict[str, int] = {"msk": 8, "spb": 4, "ekb": 4}
+
+# Deterministic branch order (KNOWN_BRANCHES is an unordered frozenset).
+_ALL_BRANCHES: tuple[str, ...] = ("msk", "spb", "ekb")
+
+
+def seed_node_baseline(conn: duckdb.DuckDBPyConnection, config: NodeConfig) -> None:
+    """Seed the per-branch baseline. Center seeds all three branches; an edge
+    seeds only its own; standalone is a no-op. Idempotent: clears any prior
+    baseline then re-lays the fixed set, so a restart yields the same baseline."""
+    if config.is_standalone or config.branch is None:
+        return
+
+    branches = _ALL_BRANCHES if config.is_center else (config.branch,)
+
+    conn.execute("BEGIN")
+    try:
+        conn.execute("DELETE FROM pipeline_events WHERE topic = ?", [NODE_BASELINE_TOPIC])
+        now = datetime.now(UTC)
+        for branch in branches:
+            for i in range(BASELINE_ROWS_BY_BRANCH[branch]):
+                conn.execute(
+                    """
+                    INSERT INTO pipeline_events (
+                        event_id, topic, tenant_id, entity_id, event_type,
+                        latency_ms, processed_at, branch
+                    )
+                    VALUES (?, ?, 'default', NULL, 'node.baseline', NULL, ?, ?)
+                    """,
+                    [f"node-baseline-{branch}-{i}", NODE_BASELINE_TOPIC, now, branch],
+                )
+        conn.execute("COMMIT")
+    except Exception:
+        conn.execute("ROLLBACK")
+        raise
diff --git a/tests/integration/test_node_topology.py b/tests/integration/test_node_topology.py
index e1d5f27..8472900 100644
--- a/tests/integration/test_node_topology.py
+++ b/tests/integration/test_node_topology.py
@@ -17,6 +17,7 @@
 
 from src.ingestion.producers.event_producer import generate_order
 from src.serving.api.main import app
+from src.serving.node.seed import BASELINE_ROWS_BY_BRANCH, NODE_BASELINE_TOPIC
 
 pytestmark = pytest.mark.integration
 
@@ -144,6 +145,30 @@ def test_center_boot_has_no_emitter(center_client: TestClient) -> None:
     assert getattr(center_client.app.state, "node_emitter_task", None) is None
 
 
+# --- N6: branch-scoped baseline seed --------------------------------------
+
+
+def _baseline_counts(client: TestClient) -> dict[str, int]:
+    rows = client.app.state.query_engine._conn.execute(
+        "SELECT branch, COUNT(*) FROM pipeline_events WHERE topic = ? GROUP BY branch",
+        [NODE_BASELINE_TOPIC],
+    ).fetchall()
+    return {str(branch): int(count) for branch, count in rows}
+
+
+def test_center_boot_seeds_all_branch_baseline(center_client: TestClient) -> None:
+    assert _baseline_counts(center_client) == BASELINE_ROWS_BY_BRANCH
+
+
+def test_edge_boot_seeds_only_its_branch(edge_client: TestClient) -> None:
+    # edge_client is the spb edge (emitter off).
+    assert _baseline_counts(edge_client) == {"spb": BASELINE_ROWS_BY_BRANCH["spb"]}
+
+
+def test_standalone_boot_seeds_no_branch_baseline(standalone_client: TestClient) -> None:
+    assert _baseline_counts(standalone_client) == {}
+
+
 # --- N3 / N10: bearer auth, not the demo-key ------------------------------
 
 
diff --git a/tests/unit/test_node_seed.py b/tests/unit/test_node_seed.py
new file mode 100644
index 0000000..107fc60
--- /dev/null
+++ b/tests/unit/test_node_seed.py
@@ -0,0 +1,82 @@
+"""Branch-scoped baseline seed — determinism + scoping (ADR 0012 §7, N6/N11)."""
+
+from __future__ import annotations
+
+import duckdb
+
+from src.processing.local_pipeline import _ensure_tables
+from src.serving.node.config import NodeConfig
+from src.serving.node.seed import (
+    BASELINE_ROWS_BY_BRANCH,
+    NODE_BASELINE_TOPIC,
+    seed_node_baseline,
+)
+
+_TOKEN = "seed-node-token"  # noqa: S105 — test fixture, not a real secret
+
+
+def _conn() -> duckdb.DuckDBPyConnection:
+    conn = duckdb.connect(":memory:")
+    _ensure_tables(conn)
+    return conn
+
+
+def _counts_by_branch(conn: duckdb.DuckDBPyConnection) -> dict[str, int]:
+    rows = conn.execute(
+        "SELECT branch, COUNT(*) FROM pipeline_events WHERE topic = ? GROUP BY branch",
+        [NODE_BASELINE_TOPIC],
+    ).fetchall()
+    return {str(branch): int(count) for branch, count in rows}
+
+
+def _snapshot(conn: duckdb.DuckDBPyConnection) -> list[tuple[str, str]]:
+    rows = conn.execute(
+        "SELECT event_id, branch FROM pipeline_events WHERE topic = ? ORDER BY event_id",
+        [NODE_BASELINE_TOPIC],
+    ).fetchall()
+    return [(str(a), str(b)) for a, b in rows]
+
+
+def test_standalone_seeds_no_baseline() -> None:
+    conn = _conn()
+    seed_node_baseline(conn, NodeConfig(role="standalone"))
+    assert _counts_by_branch(conn) == {}
+
+
+def test_center_seeds_all_branches() -> None:
+    conn = _conn()
+    seed_node_baseline(conn, NodeConfig(role="center", branch="msk", token=_TOKEN))
+    assert _counts_by_branch(conn) == BASELINE_ROWS_BY_BRANCH
+
+
+def test_edge_seeds_only_its_branch() -> None:
+    conn = _conn()
+    seed_node_baseline(
+        conn,
+        NodeConfig(role="edge", branch="spb", center_url="https://c", token=_TOKEN),
+    )
+    assert _counts_by_branch(conn) == {"spb": BASELINE_ROWS_BY_BRANCH["spb"]}
+
+
+def test_reseed_is_deterministic() -> None:
+    # N11: re-seeding (the restart analog) yields the same baseline, never a
+    # doubled one.
+    conn = _conn()
+    config = NodeConfig(role="center", branch="msk", token=_TOKEN)
+    seed_node_baseline(conn, config)
+    first = _snapshot(conn)
+
+    seed_node_baseline(conn, config)
+    assert _snapshot(conn) == first
+
+
+def test_baseline_topic_is_disjoint_from_live_topics() -> None:
+    # The baseline must not masquerade as a live validated/deadletter event.
+    conn = _conn()
+    seed_node_baseline(conn, NodeConfig(role="center", branch="msk", token=_TOKEN))
+    live = conn.execute(
+        "SELECT COUNT(*) FROM pipeline_events "
+        "WHERE topic IN ('events.validated', 'events.deadletter')"
+    ).fetchone()
+    assert live is not None
+    assert live[0] == 0

From 21bb4cfa854193424e8d51f063389b3b9ab7c050 Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 12:09:34 +0300
Subject: [PATCH 5/6] feat(node): center cross-branch view + graceful
 degradation (F2 step 5)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

GET /v1/node/branches (center-only, 404 off-center). Composes a per-branch
summary purely from the pipeline_events journal — no new store: seeded
baseline, live delta this lifecycle, and last-seen timestamp. A silent branch
(asleep / just booted) shows baseline + null last-seen + a waking status,
never an error (N8) — the sleep behaviour is visible by design. Includes the
UI hint directing visitors to open a branch Space.

N8 unit + integration tests (incl. the N4 "metric moved" end: an ingested
spb event lifts the spb live delta and sets last-seen).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/serving/node/ingest.py              | 23 ++++++++
 src/serving/node/view.py                | 55 +++++++++++++++++++
 tests/integration/test_node_topology.py | 41 ++++++++++++++
 tests/unit/test_node_view.py            | 71 +++++++++++++++++++++++++
 4 files changed, 190 insertions(+)
 create mode 100644 src/serving/node/view.py
 create mode 100644 tests/unit/test_node_view.py

diff --git a/src/serving/node/ingest.py b/src/serving/node/ingest.py
index 1312ceb..33c2ab0 100644
--- a/src/serving/node/ingest.py
+++ b/src/serving/node/ingest.py
@@ -27,6 +27,7 @@
 from starlette.concurrency import run_in_threadpool
 
 from src.processing.local_pipeline import _process_event
+from src.serving.node.view import CROSS_BRANCH_HINT, cross_branch_summary
 
 logger = structlog.get_logger()
 
@@ -152,3 +153,25 @@ def _apply() -> tuple[int, int, int]:
             "duplicates": duplicates,
         },
     )
+
+
+@router.get("/branches", include_in_schema=False, response_model=None)
+async def cross_branch_view(request: Request) -> JSONResponse:
+    """Center's cross-branch summary (ADR 0012 §9). Center-only (404 off-center);
+    a normal demo read (public demo-key), degrading gracefully on silent
+    branches (N8)."""
+    config = request.app.state.node_config
+    if not config.is_center:
+        return JSONResponse(status_code=404, content={"detail": "Not Found"})
+
+    query_engine = request.app.state.query_engine
+
+    def _read() -> list[dict]:
+        with query_engine._read_connection() as conn:
+            return cross_branch_summary(conn)
+
+    branches = await run_in_threadpool(_read)
+    return JSONResponse(
+        status_code=200,
+        content={"branches": branches, "hint": CROSS_BRANCH_HINT},
+    )
diff --git a/src/serving/node/view.py b/src/serving/node/view.py
new file mode 100644
index 0000000..4fa6c81
--- /dev/null
+++ b/src/serving/node/view.py
@@ -0,0 +1,55 @@
+"""Center cross-branch summary from the journal (ADR 0012 §9, N8).
+
+A pure read over ``pipeline_events`` — no new store. For every known branch it
+reports the seeded baseline, the live delta accrued this lifecycle, and a
+last-seen timestamp. A branch that has sent nothing (asleep or just booted)
+shows baseline + null last-seen + a ``waking`` status, **never** an error — the
+sleep behaviour is visible by design.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+import duckdb
+
+from src.serving.node.seed import NODE_BASELINE_TOPIC
+
+# All branches the center reports on, in a stable display order (msk hub first).
+CROSS_BRANCH_ORDER: tuple[str, ...] = ("msk", "spb", "ekb")
+
+CROSS_BRANCH_HINT = "open a branch Space to see its events flow to the hub"
+
+
+def cross_branch_summary(conn: duckdb.DuckDBPyConnection) -> list[dict]:
+    """Per-branch baseline / live-delta / last-seen, one row per known branch.
+
+    Silent branches are included with zero delta and a null last-seen (N8), so
+    the view degrades gracefully rather than dropping or erroring on them."""
+    baseline_rows = conn.execute(
+        "SELECT branch, COUNT(*) FROM pipeline_events WHERE topic = ? AND branch IS NOT NULL "
+        "GROUP BY branch",
+        [NODE_BASELINE_TOPIC],
+    ).fetchall()
+    baseline = {str(branch): int(count) for branch, count in baseline_rows}
+
+    live_rows = conn.execute(
+        "SELECT branch, COUNT(*), MAX(processed_at) FROM pipeline_events "
+        "WHERE topic IN ('events.validated', 'events.deadletter') AND branch IS NOT NULL "
+        "GROUP BY branch"
+    ).fetchall()
+    live = {str(branch): (int(count), last_seen) for branch, count, last_seen in live_rows}
+
+    summary = []
+    for branch in CROSS_BRANCH_ORDER:
+        live_delta, last_seen = live.get(branch, (0, None))
+        summary.append(
+            {
+                "branch": branch,
+                "baseline": baseline.get(branch, 0),
+                "live_delta": live_delta,
+                "last_seen": last_seen.isoformat() if isinstance(last_seen, datetime) else None,
+                "status": "live" if live_delta > 0 else "waking",
+            }
+        )
+    return summary
diff --git a/tests/integration/test_node_topology.py b/tests/integration/test_node_topology.py
index 8472900..20b8400 100644
--- a/tests/integration/test_node_topology.py
+++ b/tests/integration/test_node_topology.py
@@ -169,6 +169,47 @@ def test_standalone_boot_seeds_no_branch_baseline(standalone_client: TestClient)
     assert _baseline_counts(standalone_client) == {}
 
 
+# --- N8: center cross-branch view + graceful degradation ------------------
+
+
+def test_center_cross_branch_view_lists_all_branches(center_client: TestClient) -> None:
+    resp = center_client.get("/v1/node/branches", headers={"X-API-Key": "demo-key"})
+    assert resp.status_code == 200
+    data = resp.json()
+    by_branch = {row["branch"]: row for row in data["branches"]}
+    assert set(by_branch) == {"msk", "spb", "ekb"}
+    assert by_branch["msk"]["baseline"] == BASELINE_ROWS_BY_BRANCH["msk"]
+    # N8: no live events yet — silent branches are waking, not errors.
+    assert by_branch["spb"]["status"] == "waking"
+    assert by_branch["spb"]["last_seen"] is None
+    assert data["hint"]
+
+
+def test_center_cross_branch_view_reflects_ingested_event(center_client: TestClient) -> None:
+    # N4 (metric moved): an edge->center event lifts the spb live delta.
+    event = _order_event()
+    center_client.post(
+        "/v1/node/events", json={"origin_branch": "spb", "events": [event]}, headers=_AUTH
+    )
+    resp = center_client.get("/v1/node/branches", headers={"X-API-Key": "demo-key"})
+    by_branch = {row["branch"]: row for row in resp.json()["branches"]}
+    assert by_branch["spb"]["live_delta"] >= 1
+    assert by_branch["spb"]["last_seen"] is not None
+    assert by_branch["spb"]["status"] == "live"
+    # The hub's own branch stays baseline-only (it aggregates, does not emit).
+    assert by_branch["msk"]["status"] == "waking"
+
+
+def test_edge_cross_branch_view_is_404(edge_client: TestClient) -> None:
+    resp = edge_client.get("/v1/node/branches", headers={"X-API-Key": "demo-key"})
+    assert resp.status_code == 404
+
+
+def test_standalone_cross_branch_view_is_404(standalone_client: TestClient) -> None:
+    resp = standalone_client.get("/v1/node/branches", headers={"X-API-Key": "demo-key"})
+    assert resp.status_code == 404
+
+
 # --- N3 / N10: bearer auth, not the demo-key ------------------------------
 
 
diff --git a/tests/unit/test_node_view.py b/tests/unit/test_node_view.py
new file mode 100644
index 0000000..38e3327
--- /dev/null
+++ b/tests/unit/test_node_view.py
@@ -0,0 +1,71 @@
+"""Cross-branch summary — graceful degradation over the journal (ADR 0012 §9, N8)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import duckdb
+
+from src.processing.local_pipeline import _ensure_tables
+from src.serving.node.config import NodeConfig
+from src.serving.node.seed import BASELINE_ROWS_BY_BRANCH, seed_node_baseline
+from src.serving.node.view import CROSS_BRANCH_ORDER, cross_branch_summary
+
+_TOKEN = "view-node-token"  # noqa: S105 — test fixture, not a real secret
+
+
+def _conn() -> duckdb.DuckDBPyConnection:
+    conn = duckdb.connect(":memory:")
+    _ensure_tables(conn)
+    return conn
+
+
+def _add_live(conn: duckdb.DuckDBPyConnection, branch: str, event_id: str) -> None:
+    conn.execute(
+        """
+        INSERT INTO pipeline_events (
+            event_id, topic, tenant_id, entity_id, event_type, latency_ms, processed_at, branch
+        )
+        VALUES (?, 'events.validated', 'default', NULL, 'order.created', 5, ?, ?)
+        """,
+        [event_id, datetime.now(UTC), branch],
+    )
+
+
+def test_summary_lists_every_known_branch_even_when_silent() -> None:
+    conn = _conn()
+    seed_node_baseline(conn, NodeConfig(role="center", branch="msk", token=_TOKEN))
+
+    summary = cross_branch_summary(conn)
+
+    assert [row["branch"] for row in summary] == list(CROSS_BRANCH_ORDER)
+    for row in summary:
+        # N8: silent branches degrade to waking + null last-seen, never dropped.
+        assert row["baseline"] == BASELINE_ROWS_BY_BRANCH[row["branch"]]
+        assert row["live_delta"] == 0
+        assert row["last_seen"] is None
+        assert row["status"] == "waking"
+
+
+def test_summary_reflects_live_delta_and_last_seen() -> None:
+    conn = _conn()
+    seed_node_baseline(conn, NodeConfig(role="center", branch="msk", token=_TOKEN))
+    _add_live(conn, "spb", "evt-live-1")
+    _add_live(conn, "spb", "evt-live-2")
+
+    by_branch = {row["branch"]: row for row in cross_branch_summary(conn)}
+
+    assert by_branch["spb"]["live_delta"] == 2
+    assert by_branch["spb"]["last_seen"] is not None
+    assert by_branch["spb"]["status"] == "live"
+    # A branch with no live events stays waking.
+    assert by_branch["ekb"]["status"] == "waking"
+    assert by_branch["ekb"]["last_seen"] is None
+
+
+def test_summary_on_empty_journal_never_errors() -> None:
+    # No baseline, no live events (a just-booted / fully asleep view).
+    conn = _conn()
+    summary = cross_branch_summary(conn)
+    assert [row["branch"] for row in summary] == list(CROSS_BRANCH_ORDER)
+    assert all(row["baseline"] == 0 and row["status"] == "waking" for row in summary)

From f4e530e2e7d8b95291fa65f20fbb7ede50febc9e Mon Sep 17 00:00:00 2001
From: JuliaEdom <uedomskikh@gmail.com>
Date: Sat, 4 Jul 2026 12:13:29 +0300
Subject: [PATCH 6/6] docs(node): per-role Space READMEs + deploy runbook (F2
 step 6 artifacts)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Per-role Hugging Face Space artifacts for the three-node deploy (ADR 0012 §11):
center/edge-spb/edge-ekb READMEs (Space frontmatter + role env matrix + the
honesty boundary stated plainly) and a DEPLOY.md runbook (same image x3, env in
Space Settings, center-first bring-up, verify-live §12). The shared image is the
existing deploy/hf-space/Dockerfile.

The actual HF publish + live-verify remain the OWNER GATE — these are the
committed artifacts only. Frontmatter guard test included.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 deploy/hf-space/three-node/DEPLOY.md          | 89 +++++++++++++++++++
 deploy/hf-space/three-node/center/README.md   | 76 ++++++++++++++++
 deploy/hf-space/three-node/edge-ekb/README.md | 65 ++++++++++++++
 deploy/hf-space/three-node/edge-spb/README.md | 65 ++++++++++++++
 docs/three-node-demo-topology.md              |  7 +-
 tests/unit/test_node_deploy_artifacts.py      | 55 ++++++++++++
 6 files changed, 355 insertions(+), 2 deletions(-)
 create mode 100644 deploy/hf-space/three-node/DEPLOY.md
 create mode 100644 deploy/hf-space/three-node/center/README.md
 create mode 100644 deploy/hf-space/three-node/edge-ekb/README.md
 create mode 100644 deploy/hf-space/three-node/edge-spb/README.md
 create mode 100644 tests/unit/test_node_deploy_artifacts.py

diff --git a/deploy/hf-space/three-node/DEPLOY.md b/deploy/hf-space/three-node/DEPLOY.md
new file mode 100644
index 0000000..c8be132
--- /dev/null
+++ b/deploy/hf-space/three-node/DEPLOY.md
@@ -0,0 +1,89 @@
+# Three-node demo — deploy runbook (owner-gated)
+
+Publishes the three-node topology (ADR 0012 / `docs/three-node-demo-topology.md`)
+as **three** Hugging Face Docker Spaces under the `liovina` account:
+
+| Node   | Space                        | Role env       | Branch |
+|--------|------------------------------|----------------|--------|
+| Center | `liovina/agentflow-center`   | `center`       | `msk`  |
+| Edge 1 | `liovina/agentflow-edge-spb` | `edge`         | `spb`  |
+| Edge 2 | `liovina/agentflow-edge-ekb` | `edge`         | `ekb`  |
+
+All three build the **same image** — the existing `deploy/hf-space/Dockerfile`,
+tracking `main`. Role is pure environment set in each Space's **Settings**, not
+in the image. The standalone `liovina/agentflow-demo` Space is unaffected; this
+set is additive.
+
+Each Space repo root gets two files: the shared `Dockerfile` and that node's
+`README.md` (from `center/`, `edge-spb/`, `edge-ekb/` here).
+
+## Deploy (outward — OWNER GATE)
+
+Creating/pushing public Spaces under `liovina` is an external publish and is an
+**owner gate**. The HF token lives in `D:/VacancyRadar/.env` (`HF_TOKEN`); never
+print it. Bring-up order is **center first** — the edges need its URL live and a
+matching `AGENTFLOW_NODE_TOKEN`.
+
+Pick one strong shared token value for `AGENTFLOW_NODE_TOKEN` (same on all three;
+store as a Space **secret**, never a variable, never logged).
+
+```bash
+# For each of: agentflow-center, agentflow-edge-spb, agentflow-edge-ekb
+NAME=agentflow-center            # then edge-spb, edge-ekb
+ROLEDIR=center                   # then edge-spb, edge-ekb
+
+# 1. Create the Space (one-time)
+huggingface-cli repo create "$NAME" --repo-type space --space_sdk docker
+
+# 2. Push the shared Dockerfile + this node's README to the Space repo root
+git clone "https://huggingface.co/spaces/liovina/$NAME" "/tmp/$NAME"
+cp deploy/hf-space/Dockerfile "/tmp/$NAME/Dockerfile"
+cp "deploy/hf-space/three-node/$ROLEDIR/README.md" "/tmp/$NAME/README.md"
+cd "/tmp/$NAME" && git add Dockerfile README.md \
+  && git commit -m "AgentFlow $NAME (three-node demo)" && git push
+```
+
+Then, in each Space's **Settings → Variables and secrets**, set the environment
+from that node's README table:
+
+- **Center:** `AGENTFLOW_NODE_ROLE=center`, `AGENTFLOW_NODE_BRANCH=msk`,
+  `AGENTFLOW_DEMO_MODE=true`, `AGENTFLOW_SEED_ON_BOOT=true`; secret
+  `AGENTFLOW_NODE_TOKEN`.
+- **Edge spb / ekb:** `AGENTFLOW_NODE_ROLE=edge`,
+  `AGENTFLOW_NODE_BRANCH=spb|ekb`,
+  `AGENTFLOW_NODE_CENTER_URL=https://liovina-agentflow-center.hf.space`,
+  `AGENTFLOW_DEMO_MODE=true`, `AGENTFLOW_SEED_ON_BOOT=true`; secret
+  `AGENTFLOW_NODE_TOKEN` (same value as the center).
+
+A misconfigured edge (no center URL or no token) fails its boot fast by design.
+
+## Verify live (§12 of the build contract)
+
+```bash
+CENTER=https://liovina-agentflow-center.hf.space
+SPB=https://liovina-agentflow-edge-spb.hf.space
+EKB=https://liovina-agentflow-edge-ekb.hf.space
+TOKEN=...   # the shared node token, from your secret store — never commit it
+
+# 1. All three healthy
+curl -fsS $CENTER/v1/health && curl -fsS $SPB/v1/health && curl -fsS $EKB/v1/health
+
+# 2. Public demo-key cannot push (403 — demo-guard holds)
+curl -i -X POST -H "X-API-Key: demo-key" -H "Content-Type: application/json" \
+  -d '{"origin_branch":"spb","events":[]}' $CENTER/v1/node/events
+
+# 3. Cross-branch view before any live event — silent branches show "waking"
+curl -fsS -H "X-API-Key: demo-key" $CENTER/v1/node/branches
+
+# 4. Open an edge (wakes it; its generator emits to the hub), wait, then re-read
+#    the center view — that branch's live_delta is non-zero and last_seen is set.
+curl -fsS $SPB/v1/health
+sleep 30
+curl -fsS -H "X-API-Key: demo-key" $CENTER/v1/node/branches
+```
+
+## Refresh after a repo change
+
+Each Space tracks `main` (`ARG AGENTFLOW_REF=main`). Trigger a **Factory
+rebuild** from the Space UI after a merge, or bump `AGENTFLOW_REF` to a tag for a
+pinned demo.
diff --git a/deploy/hf-space/three-node/center/README.md b/deploy/hf-space/three-node/center/README.md
new file mode 100644
index 0000000..1d676ca
--- /dev/null
+++ b/deploy/hf-space/three-node/center/README.md
@@ -0,0 +1,76 @@
+---
+title: AgentFlow — Center (msk hub)
+colorFrom: indigo
+colorTo: gray
+sdk: docker
+app_port: 8000
+pinned: false
+---
+
+# AgentFlow — Center node (`msk` hub)
+
+The **hub** of the three-node demo (ADR 0012). It runs the serving API in demo
+mode and additionally accepts operational events pushed by the regional edge
+branches over HTTPS, aggregating a **live cross-branch view**. Built from the
+public source at
+[github.com/brownjuly2003-code/agentflow](https://github.com/brownjuly2003-code/agentflow),
+tracking `main` — the same image as the edges, differentiated only by
+environment.
+
+The other two nodes:
+
+- **Edge `spb`** — https://liovina-agentflow-edge-spb.hf.space
+- **Edge `ekb`** — https://liovina-agentflow-edge-ekb.hf.space
+
+## What this node adds over the standalone demo
+
+- `GET /v1/node/branches` — the cross-branch summary: per branch a seeded
+  baseline, the live delta accrued this lifecycle, and a last-seen timestamp. A
+  branch that has sent nothing shows a `waking` status — open its Space to see
+  its events flow here.
+- `POST /v1/node/events` — token-authenticated ingest for edge→center events
+  (not the public `demo-key`; internal node-to-node, hidden from `/docs`).
+
+## What this demo shows — and what it does NOT
+
+This is a demo of **branch distribution / event federation**: distinct
+geographic branch nodes emit operational events to a hub that aggregates a
+cross-branch view. Stated plainly so the artifact never over-claims:
+
+- **Transport:** direct HTTPS `POST` stands in for Kafka → Flink. The event
+  *semantics* are identical (the same canonical events drive the same journal
+  and metric freshness); only the transport differs, because Spaces cannot run
+  Kafka.
+- **State:** each node is single-replica with an **embedded** control plane —
+  the correct single-replica profile, **not** the multi-replica PostgreSQL
+  scale profile (that is the Kubernetes cutover, verified separately). No
+  database spans the Spaces.
+- **Liveness:** the federated live layer is **ephemeral**. Free Spaces have
+  non-persistent disk and sleep after 48 h of inactivity; on a restart the hub
+  re-seeds a deterministic baseline and the live layer resets to zero.
+
+## Environment (set in Space **Settings**, not the image)
+
+| Variable                | Value                          | Kind     |
+|-------------------------|--------------------------------|----------|
+| `AGENTFLOW_NODE_ROLE`   | `center`                       | variable |
+| `AGENTFLOW_NODE_BRANCH` | `msk`                          | variable |
+| `AGENTFLOW_NODE_TOKEN`  | shared node token              | **secret** |
+| `AGENTFLOW_DEMO_MODE`   | `true`                         | variable |
+| `AGENTFLOW_SEED_ON_BOOT`| `true`                         | variable |
+
+## Try it
+
+```bash
+SPACE=https://liovina-agentflow-center.hf.space
+
+# Liveness
+curl -fsS $SPACE/v1/health
+
+# Cross-branch view (public demo key)
+curl -fsS -H "X-API-Key: demo-key" $SPACE/v1/node/branches
+
+# Public callers cannot push events — the node token is required (403)
+curl -i -X POST -H "X-API-Key: demo-key" -H "Content-Type: application/json" \
+  -d '{"origin_branch":"spb","events":[]}' $SPACE/v1/node/events
+```
diff --git a/deploy/hf-space/three-node/edge-ekb/README.md b/deploy/hf-space/three-node/edge-ekb/README.md
new file mode 100644
index 0000000..cf3e5a6
--- /dev/null
+++ b/deploy/hf-space/three-node/edge-ekb/README.md
@@ -0,0 +1,65 @@
+---
+title: AgentFlow — Edge (ekb)
+colorFrom: gray
+colorTo: indigo
+sdk: docker
+app_port: 8000
+pinned: false
+---
+
+# AgentFlow — Edge node (`ekb`)
+
+A regional-warehouse **edge** of the three-node demo (ADR 0012). It runs the
+serving API in demo mode over its own `ekb` branch slice and, while awake, emits
+operational events over HTTPS to the **center hub**, whose cross-branch view
+reflects them live. Same image as the hub and the other edge, differentiated
+only by environment; built from
+[github.com/brownjuly2003-code/agentflow](https://github.com/brownjuly2003-code/agentflow),
+tracking `main`.
+
+- **Center hub** — https://liovina-agentflow-center.hf.space (open it to see the
+  cross-branch view)
+- **Edge `spb`** — https://liovina-agentflow-edge-spb.hf.space
+
+## How the federation works
+
+A slow background generator produces the same canonical events the in-process
+pipeline makes; each is applied to this edge's own read surface **and** forwarded
+to the hub's `POST /v1/node/events`. Push-on-activity is what makes the sleep
+choreography self-healing: if the hub is asleep, the first forwarded event wakes
+it; a cold hub is tolerated (bounded retries, then drop) so this page stays live
+regardless.
+
+## What this demo shows — and what it does NOT
+
+Branch distribution / event federation, **not** horizontal scaling of one node:
+
+- **Transport:** HTTPS `POST` stands in for Kafka → Flink; event semantics are
+  identical, only the transport differs (Spaces cannot run Kafka).
+- **State:** single-replica, embedded control plane — **not** the multi-replica
+  PostgreSQL scale profile (the Kubernetes cutover, verified separately).
+- **Liveness:** ephemeral. Free Spaces sleep after 48 h and have non-persistent
+  disk; a restart re-seeds a deterministic baseline and resets the live layer.
+
+## Environment (set in Space **Settings**, not the image)
+
+| Variable                   | Value                                         | Kind     |
+|----------------------------|-----------------------------------------------|----------|
+| `AGENTFLOW_NODE_ROLE`      | `edge`                                        | variable |
+| `AGENTFLOW_NODE_BRANCH`    | `ekb`                                         | variable |
+| `AGENTFLOW_NODE_CENTER_URL`| `https://liovina-agentflow-center.hf.space`   | variable |
+| `AGENTFLOW_NODE_TOKEN`     | shared node token (same value as the hub)     | **secret** |
+| `AGENTFLOW_DEMO_MODE`      | `true`                                        | variable |
+| `AGENTFLOW_SEED_ON_BOOT`   | `true`                                        | variable |
+
+## Try it
+
+```bash
+SPACE=https://liovina-agentflow-edge-ekb.hf.space
+
+# Liveness (also wakes the edge, which starts emitting to the hub)
+curl -fsS $SPACE/v1/health
+
+# This edge's own read surface (public demo key)
+curl -fsS -H "X-API-Key: demo-key" $SPACE/v1/entity/order/ORD-20260404-1001
+```
diff --git a/deploy/hf-space/three-node/edge-spb/README.md b/deploy/hf-space/three-node/edge-spb/README.md
new file mode 100644
index 0000000..b31010b
--- /dev/null
+++ b/deploy/hf-space/three-node/edge-spb/README.md
@@ -0,0 +1,65 @@
+---
+title: AgentFlow — Edge (spb)
+colorFrom: gray
+colorTo: indigo
+sdk: docker
+app_port: 8000
+pinned: false
+---
+
+# AgentFlow — Edge node (`spb`)
+
+A regional-warehouse **edge** of the three-node demo (ADR 0012). It runs the
+serving API in demo mode over its own `spb` branch slice and, while awake, emits
+operational events over HTTPS to the **center hub**, whose cross-branch view
+reflects them live. Same image as the hub and the other edge, differentiated
+only by environment; built from
+[github.com/brownjuly2003-code/agentflow](https://github.com/brownjuly2003-code/agentflow),
+tracking `main`.
+
+- **Center hub** — https://liovina-agentflow-center.hf.space (open it to see the
+  cross-branch view)
+- **Edge `ekb`** — https://liovina-agentflow-edge-ekb.hf.space
+
+## How the federation works
+
+A slow background generator produces the same canonical events the in-process
+pipeline makes; each is applied to this edge's own read surface **and** forwarded
+to the hub's `POST /v1/node/events`. Push-on-activity is what makes the sleep
+choreography self-healing: if the hub is asleep, the first forwarded event wakes
+it; a cold hub is tolerated (bounded retries, then drop) so this page stays live
+regardless.
+
+## What this demo shows — and what it does NOT
+
+Branch distribution / event federation, **not** horizontal scaling of one node:
+
+- **Transport:** HTTPS `POST` stands in for Kafka → Flink; event semantics are
+  identical, only the transport differs (Spaces cannot run Kafka).
+- **State:** single-replica, embedded control plane — **not** the multi-replica
+  PostgreSQL scale profile (the Kubernetes cutover, verified separately).
+- **Liveness:** ephemeral. Free Spaces sleep after 48 h and have non-persistent
+  disk; a restart re-seeds a deterministic baseline and resets the live layer.
+
+## Environment (set in Space **Settings**, not the image)
+
+| Variable                   | Value                                         | Kind     |
+|----------------------------|-----------------------------------------------|----------|
+| `AGENTFLOW_NODE_ROLE`      | `edge`                                        | variable |
+| `AGENTFLOW_NODE_BRANCH`    | `spb`                                         | variable |
+| `AGENTFLOW_NODE_CENTER_URL`| `https://liovina-agentflow-center.hf.space`   | variable |
+| `AGENTFLOW_NODE_TOKEN`     | shared node token (same value as the hub)     | **secret** |
+| `AGENTFLOW_DEMO_MODE`      | `true`                                        | variable |
+| `AGENTFLOW_SEED_ON_BOOT`   | `true`                                        | variable |
+
+## Try it
+
+```bash
+SPACE=https://liovina-agentflow-edge-spb.hf.space
+
+# Liveness (also wakes the edge, which starts emitting to the hub)
+curl -fsS $SPACE/v1/health
+
+# This edge's own read surface (public demo key)
+curl -fsS -H "X-API-Key: demo-key" $SPACE/v1/entity/order/ORD-20260404-1001
+```
diff --git a/docs/three-node-demo-topology.md b/docs/three-node-demo-topology.md
index 7def3d5..4ebe3fc 100644
--- a/docs/three-node-demo-topology.md
+++ b/docs/three-node-demo-topology.md
@@ -179,8 +179,11 @@ default** behind a flag - the core design is push-only (ADR 0012 Decision 5).
 
 ## 11. Deploy runbook skeleton (owner gate)
 
-Mirrors `deploy/hf-space/DEPLOY.md` x3. `HF_TOKEN` from `D:/VacancyRadar/.env` - never
-printed. For each of `agentflow-center`, `agentflow-edge-spb`, `agentflow-edge-ekb`:
+The concrete runbook and per-role Space READMEs live in
+`deploy/hf-space/three-node/` (`DEPLOY.md` + `center/`, `edge-spb/`, `edge-ekb/`
+READMEs; the shared image is the existing `deploy/hf-space/Dockerfile`).
+`HF_TOKEN` from `D:/VacancyRadar/.env` - never printed. For each of
+`agentflow-center`, `agentflow-edge-spb`, `agentflow-edge-ekb`:
 
 ```bash
 # 1. Create (one-time)
diff --git a/tests/unit/test_node_deploy_artifacts.py b/tests/unit/test_node_deploy_artifacts.py
new file mode 100644
index 0000000..6b85f0a
--- /dev/null
+++ b/tests/unit/test_node_deploy_artifacts.py
@@ -0,0 +1,55 @@
+"""Guard the three-node deploy artifacts (ADR 0012 §11).
+
+A malformed Space README frontmatter is a silent deploy failure (the HF Space
+build rejects it), so these check the deploy-critical shape and that each
+per-role README documents the environment that role needs.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+_ROOT = Path(__file__).resolve().parents[2]
+_THREE_NODE = _ROOT / "deploy" / "hf-space" / "three-node"
+
+
+def _frontmatter(readme: Path) -> dict:
+    text = readme.read_text(encoding="utf-8")
+    assert text.startswith("---\n"), f"{readme} is missing YAML frontmatter"
+    _, block, _body = text.split("---\n", 2)
+    return yaml.safe_load(block)
+
+
+def test_deploy_runbook_exists() -> None:
+    assert (_THREE_NODE / "DEPLOY.md").is_file()
+
+
+def test_each_role_readme_has_valid_space_frontmatter() -> None:
+    for role in ("center", "edge-spb", "edge-ekb"):
+        readme = _THREE_NODE / role / "README.md"
+        assert readme.is_file(), f"missing {readme}"
+        meta = _frontmatter(readme)
+        assert meta["sdk"] == "docker"
+        assert meta["app_port"] == 8000
+        assert meta["title"]
+
+
+def test_center_readme_documents_center_role() -> None:
+    text = (_THREE_NODE / "center" / "README.md").read_text(encoding="utf-8")
+    assert "AGENTFLOW_NODE_ROLE" in text
+    assert "`center`" in text
+    assert "`msk`" in text
+    # The honesty boundary must be stated so the artifact never over-claims.
+    assert "scale profile" in text
+    assert "ephemeral" in text.lower()
+
+
+def test_edge_readmes_document_edge_role_and_center_url() -> None:
+    for role, branch in (("edge-spb", "spb"), ("edge-ekb", "ekb")):
+        text = (_THREE_NODE / role / "README.md").read_text(encoding="utf-8")
+        assert "AGENTFLOW_NODE_ROLE" in text
+        assert "`edge`" in text
+        assert f"`{branch}`" in text
+        assert "AGENTFLOW_NODE_CENTER_URL" in text