fix: persist agent name on sub-session nodes across a delivery-order race (v6.7.4)

context_intelligence_server/services.py

@@ -317,6 +317,16 @@ async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> No
             # guarantees an already-set value is never clobbered.
             if data.get("working_dir") and not existing.get("working_dir"):
                 stub_data["working_dir"] = data["working_dir"]
+            # Issue #484: same populate-if-missing rule for `agent`. The agent
+            # name for a spawned sub-session arrives ONLY on the parent's
+            # delegate:agent_spawned event; the child's own session:start (no
+            # top-level agent) can create the node first, so that later parent
+            # event routinely lands HERE. Without this, the parent's `agent` was
+            # silently dropped, leaving :Session.agent empty and breaking any
+            # `WHERE s.agent = ...` query. Only writes when the event supplies it
+            # AND the node still lacks it, so an already-set value is preserved.
+            if data.get("agent") and not existing.get("agent"):
+                stub_data["agent"] = data["agent"]
             await self.graph.upsert_node(session_id, stub_data)
             self._seen_sessions.add(session_id)
             return

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "context-intelligence-server"
-version = "6.7.3"
+version = "6.7.4"
 description = "Context Intelligence Server for Amplifier"
 requires-python = ">=3.11"
 dependencies = [

tests/neo4j/test_agent_field_ordering_race.py

@@ -0,0 +1,142 @@
+"""E2E reproduction for issue #484: `agent` dropped on the child-first ordering.
+
+Runs against an ISOLATED, throwaway Neo4j container (the ``neo4j_container``
+fixture in ``tests/neo4j/conftest.py`` — random ports, ``remove=True``, torn
+down after the session). NEVER touches the production/shared store.
+
+The race
+--------
+A spawned sub-session's ``agent`` name arrives on the PARENT's
+``delegate:agent_spawned`` event (top-level ``agent``). The CHILD's own
+``session:start`` carries no top-level ``agent``. Both are processed through
+``ensure_session_node`` (pipeline step 2). ``ensure_session_node``'s "node
+already exists" (Tier-2) branch historically upserted only
+``{labels, status, session_id}`` — so when the CHILD's ``session:start`` created
+the node first, the PARENT's later ``delegate:agent_spawned`` (which DOES carry
+``{"agent": ...}``) hit the Tier-2 branch and its ``agent`` was silently dropped,
+leaving ``:Session.agent`` permanently empty.
+
+This test drives the real handlers across two independent ``Neo4jGraphStore``
+instances sharing one Neo4j (exactly the two-drainer condition), forcing the
+CHILD-first ordering deterministically, then asserts ``agent`` is persisted.
+
+RED before the services.py fix, GREEN after.
+
+Run: uv run pytest tests/neo4j/test_agent_field_ordering_race.py -v -m neo4j
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+import pytest
+
+from context_intelligence_server.handlers.data_layer_2.session import SessionHandler
+from context_intelligence_server.handlers.data_layer_3.delegation import (
+    DelegationHandler,
+)
+from context_intelligence_server.neo4j_store import (
+    Neo4jGraphStore,
+    ensure_neo4j_schema,
+)
+from context_intelligence_server.services import HookStateService
+
+pytestmark = pytest.mark.neo4j
+
+
+async def _neo4j_agent(store: Neo4jGraphStore, node_id: str) -> str | None:
+    """Read the `agent` property of a node straight from Neo4j (not the buffer)."""
+    rows = await store.execute_query(
+        "MATCH (n) WHERE n.node_id = $id AND n.workspace = $workspace "
+        "RETURN n.agent AS agent",
+        {"id": node_id, "workspace": store.workspace},
+        workspace="*",
+    )
+    return rows[0]["agent"] if rows else None
+
+
+def _ts(n: int = 0) -> str:
+    return f"2026-01-01T00:{n:02d}:00Z"
+
+
+@pytest.mark.neo4j
+class TestAgentFieldChildFirstOrdering:
+    """#484: agent must survive when the child's session:start lands first."""
+
+    async def test_child_start_before_parent_spawn_persists_agent(
+        self, neo4j_container: dict[str, Any]
+    ) -> None:
+        """CHILD session:start creates the node first (no agent); the PARENT's
+        later delegate:agent_spawned must still persist `agent` onto it."""
+        auth = (neo4j_container["user"], neo4j_container["password"])
+        bolt = neo4j_container["bolt_url"]
+        ws = f"test-agent-484-{uuid.uuid4().hex[:8]}"
+
+        from neo4j import AsyncGraphDatabase
+
+        driver = AsyncGraphDatabase.driver(bolt, auth=auth)
+        await ensure_neo4j_schema(driver)
+        await driver.close()
+
+        parent_id = f"parent-{uuid.uuid4().hex[:8]}"
+        child_id = f"child-{uuid.uuid4().hex[:8]}"
+        tool_call_id = f"tc-{uuid.uuid4().hex[:8]}"
+        expected_agent = "foundation:git-ops"
+
+        # --- CHILD's drainer resources ---
+        child_store = Neo4jGraphStore(uri=bolt, auth=auth, workspace=ws)
+        child_services = HookStateService(workspace=ws, graph_store=child_store)
+        session_handler_child = SessionHandler(child_services)
+
+        # --- PARENT's drainer resources ---
+        parent_store = Neo4jGraphStore(uri=bolt, auth=auth, workspace=ws)
+        parent_services = HookStateService(workspace=ws, graph_store=parent_store)
+        await parent_services.ensure_session_node(parent_id, {})
+        delegation_handler = DelegationHandler(parent_services)
+
+        # Step 1 — CHILD's own session:start creates the sub-session node FIRST,
+        # carrying NO top-level agent (agent name is only nested in metadata).
+        await session_handler_child(
+            "session:start",
+            {
+                "session_id": child_id,
+                "parent_id": parent_id,
+                "timestamp": _ts(1),
+                "metadata": {"agent_name": expected_agent},
+            },
+        )
+        await child_store.flush()
+
+        # Precondition: node exists in Neo4j but has no agent yet.
+        assert await _neo4j_agent(child_store, child_id) is None
+
+        # Step 2 — PARENT's delegate:agent_spawned arrives LATER, carrying the
+        # top-level agent. This is the only event that supplies `agent` to
+        # ensure_session_node, and it now hits the Tier-2 existing-node branch.
+        await delegation_handler(
+            "delegate:agent_spawned",
+            {
+                "session_id": parent_id,
+                "parent_session_id": parent_id,
+                "sub_session_id": child_id,
+                "agent": expected_agent,
+                "tool_call_id": tool_call_id,
+                "timestamp": _ts(0),
+            },
+        )
+        await parent_store.flush()
+
+        # Assert — the sub-session node carries the agent end-to-end in Neo4j.
+        verify_store = Neo4jGraphStore(uri=bolt, auth=auth, workspace=ws)
+        try:
+            actual = await _neo4j_agent(verify_store, child_id)
+            assert actual == expected_agent, (
+                f"#484 REPRODUCED: sub-session {child_id} has agent={actual!r}, "
+                f"expected {expected_agent!r}. The parent's delegate:agent_spawned "
+                f"agent value was dropped by ensure_session_node's Tier-2 branch."
+            )
+        finally:
+            await verify_store.close()
+            await child_store.close()
+            await parent_store.close()

tests/test_services.py

@@ -748,3 +748,68 @@ def test_corrupt_record_never_raises(self) -> None:
         svc = HookStateService(workspace="/ws")
         svc.restore_cursor({"dl2": "not-a-dict", "dl3": 123})  # type: ignore[dict-item]
         assert svc.data_layer_2.iteration_count == 0
+
+
+# ---------------------------------------------------------------------------
+# Issue #484: ensure_session_node must not drop `agent` on the existing-node
+# (Tier-2) branch. The `agent` value for a spawned sub-session arrives on the
+# parent's delegate:agent_spawned event ({"agent": ...}), while the child's own
+# session:start (no top-level agent) can create the node first. When the parent
+# event then hits the Tier-2 branch, `agent` must still be persisted (same
+# populate-if-missing rule already applied to working_dir).
+#
+# Two HookStateService instances share ONE GraphState to model the real
+# two-writer condition (each worker has its own cold _seen_sessions cache, so
+# the second writer genuinely reaches the graph-query Tier-2 branch rather than
+# short-circuiting on the Tier-1 warm-cache fast path).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestEnsureSessionNodeAgentField:
+    """#484 regression: the agent name survives the child-first ordering race."""
+
+    async def test_tier2_existing_node_persists_agent_from_later_event(self) -> None:
+        """Child creates the node WITHOUT agent; a later writer carrying
+        {"agent": ...} (the parent's delegate:agent_spawned) must persist it."""
+        graph = GraphState()
+        svc_child = HookStateService(graph_store=graph)
+        svc_parent = HookStateService(graph_store=graph)
+
+        # Child's session:start reaches ensure_session_node first — no top-level agent.
+        await svc_child.ensure_session_node(
+            "child-484", {"timestamp": "2026-01-01T00:00:00Z"}
+        )
+        node = await graph.get_node("child-484")
+        assert node is not None
+        # Precondition: the node exists but has no agent yet (reproduces the setup).
+        assert node.get("agent") is None
+
+        # Parent's delegate:agent_spawned arrives later, carrying the agent name.
+        # Second writer's cache is cold, so this reaches the Tier-2 existing branch.
+        await svc_parent.ensure_session_node(
+            "child-484", {"agent": "foundation:git-ops"}
+        )
+
+        node = await graph.get_node("child-484")
+        assert node is not None
+        assert node.get("agent") == "foundation:git-ops"
+
+    async def test_tier2_does_not_clobber_existing_agent(self) -> None:
+        """A later agent-less writer must NOT wipe an already-set agent."""
+        graph = GraphState()
+        svc_parent = HookStateService(graph_store=graph)
+        svc_child = HookStateService(graph_store=graph)
+
+        # Parent creates the node first, with the agent set.
+        await svc_parent.ensure_session_node(
+            "child-484b", {"agent": "foundation:git-ops"}
+        )
+        # Child's later agent-less call hits Tier-2 and must not clobber agent.
+        await svc_child.ensure_session_node(
+            "child-484b", {"timestamp": "2026-01-01T00:00:00Z"}
+        )
+
+        node = await graph.get_node("child-484b")
+        assert node is not None
+        assert node.get("agent") == "foundation:git-ops"