From 943e87c6a5fb240d0818cd9c26f6168dfe12c467 Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 24 Aug 2026 18:54:22 +0000 Subject: [PATCH] fix(services): persist `agent` on sub-session nodes across delivery-order race (#484) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ensure_session_node's Tier-2 (node-already-exists) branch dropped the `agent` value carried by the parent's delegate:agent_spawned event when the child's own session:start created the node first. Apply the same populate-if-missing rule already used for working_dir so the agent name is backfilled onto the existing node without ever clobbering an already-set value. Fixes the empty :Session.agent that silently undercounted `WHERE s.agent = ...` queries on the shared server. Also bump server version 6.7.3 -> 6.7.4. Tests (RED before fix, GREEN after): - unit: in-memory two-writer reproduction of the child-first ordering - E2E: isolated throwaway Neo4j, real Session/Delegation handlers, child session:start before parent delegate:agent_spawned; asserts agent persisted 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- context_intelligence_server/services.py | 10 ++ pyproject.toml | 2 +- tests/neo4j/test_agent_field_ordering_race.py | 142 ++++++++++++++++++ tests/test_services.py | 65 ++++++++ uv.lock | 2 +- 5 files changed, 219 insertions(+), 2 deletions(-) create mode 100644 tests/neo4j/test_agent_field_ordering_race.py diff --git a/context_intelligence_server/services.py b/context_intelligence_server/services.py index ccb154fa..355f0043 100644 --- a/context_intelligence_server/services.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 0e032375..9742a0bc 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/tests/neo4j/test_agent_field_ordering_race.py b/tests/neo4j/test_agent_field_ordering_race.py new file mode 100644 index 00000000..16ae73c9 --- /dev/null +++ b/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() diff --git a/tests/test_services.py b/tests/test_services.py index 601d19b5..3362f0d1 100644 --- a/tests/test_services.py +++ b/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" diff --git a/uv.lock b/uv.lock index ee1380f4..d3a12b52 100644 --- a/uv.lock +++ b/uv.lock @@ -233,7 +233,7 @@ wheels = [ [[package]] name = "context-intelligence-server" -version = "6.7.3" +version = "6.7.4" source = { editable = "." } dependencies = [ { name = "aiofiles" },