PR2: Session-node data quality — working_dir + agent + heal-forward

CHANGELOG.md

@@ -5,6 +5,33 @@ All notable changes to the Context Intelligence Server are recorded here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [7.1.0]
+
+### Added
+
+- **`working_dir` end-to-end on the Session node.** `EventRequest` accepts an
+  optional top-level `working_dir` envelope field (rejected only when
+  blank/whitespace-only); the events endpoint lifts it into the event data, and
+  `ensure_session_node` writes it onto the Session node. Populate-if-missing: a
+  node created before `working_dir` was known is backfilled by the first
+  subsequent event that carries a non-empty value. An already-set value is
+  never overwritten — the Neo4j write uses
+  `coalesce(n.working_dir, row.working_dir)` rather than last-write-wins.
+
+### Fixed
+
+- **`agent` persists across the delivery-order race.** The agent name for a
+  spawned sub-session arrives only on the parent's `delegate:agent_spawned`
+  event, but the child's own `session:start` can create the Session node first.
+  `ensure_session_node` now backfills `agent` on the existing-node branch with
+  the same populate-if-missing rule as `working_dir`, so it is no longer
+  silently dropped (which left `:Session.agent` empty and undercounted
+  `WHERE s.agent = ...` queries).
+- **IncompleteSession heal-forward.** A stale `IncompleteSession` marker —
+  stamped when a session's `session:end` drained before its `session:start`/
+  `session:fork` — is now stripped the moment the real start/fork is processed,
+  leaving only the correct terminal label.
+
 ## [7.0.0]
 
 ### Changed (breaking)

context_intelligence_server/handlers/data_layer_2/session.py

@@ -68,46 +68,36 @@ class SessionLabelStateMachine:
     ForkedSession > SubSession > RootSession in specificity (terminal ordering).
     """
 
+    @staticmethod
+    def _heal_forward(transition: LabelTransition) -> LabelTransition:
+        """Strip a stale IncompleteSession marker on a start/fork transition.
+
+        IncompleteSession is only correct at session:end on a bare session whose
+        start/fork was lost. Its co-occurrence with a start/fork is always the
+        out-of-order case -- a forked sub-session's session:end drained before
+        its session:start/fork, stamping the marker first. Removing an absent
+        label is a no-op, so applying this to every start/fork result is safe and
+        makes the invariant impossible to miss when a branch is added later.
+        """
+        if "IncompleteSession" in transition.remove:
+            return transition
+        return LabelTransition(
+            add=transition.add, remove=[*transition.remove, "IncompleteSession"]
+        )
+
     def classify(
         self, current_type: str | None, event: str, has_parent: bool
     ) -> LabelTransition:
         # StubSession is a plain observability marker, not part of the
         # terminal lattice; every branch assigning a real terminal label
-        # also clears it (removing an absent label is a no-op).
+        # also clears it (removing an absent label is a no-op). Start/fork
+        # results are passed through _heal_forward so a stale IncompleteSession
+        # marker is stripped on the transition (it is only correct at end).
         if event == "start":
-            if current_type in ("ForkedSession", "SubSession"):
-                return LabelTransition()
-            if current_type == "RootSession":
-                if has_parent:
-                    return LabelTransition(
-                        add=["SubSession", "SST_EVENT"],
-                        remove=["RootSession", "StubSession"],
-                    )
-                return LabelTransition()
-            # bare session (current_type is None)
-            if has_parent:
-                return LabelTransition(
-                    add=["Session", "SubSession", "SST_EVENT"],
-                    remove=["StubSession"],
-                )
-            return LabelTransition(
-                add=["RootSession", "Session", "SST_EVENT"],
-                remove=["StubSession"],
-            )
+            return self._heal_forward(self._classify_start(current_type, has_parent))
 
         if event == "fork":
-            if current_type == "ForkedSession":
-                return LabelTransition()
-            if current_type in ("RootSession", "SubSession"):
-                return LabelTransition(
-                    add=["ForkedSession", "SST_EVENT"],
-                    remove=[current_type, "StubSession"],
-                )
-            # bare session (current_type is None)
-            return LabelTransition(
-                add=["Session", "ForkedSession", "SST_EVENT"],
-                remove=["StubSession"],
-            )
+            return self._heal_forward(self._classify_fork(current_type, has_parent))
 
         if event == "end":
             if current_type is not None:
@@ -120,6 +110,43 @@ def classify(
 
         raise ValueError(f"classify() received unknown event: {event!r}")
 
+    @staticmethod
+    def _classify_start(current_type: str | None, has_parent: bool) -> LabelTransition:
+        if current_type in ("ForkedSession", "SubSession"):
+            return LabelTransition()
+        if current_type == "RootSession":
+            if has_parent:
+                return LabelTransition(
+                    add=["SubSession", "SST_EVENT"],
+                    remove=["RootSession", "StubSession"],
+                )
+            return LabelTransition()
+        # bare session (current_type is None)
+        if has_parent:
+            return LabelTransition(
+                add=["Session", "SubSession", "SST_EVENT"],
+                remove=["StubSession"],
+            )
+        return LabelTransition(
+            add=["RootSession", "Session", "SST_EVENT"],
+            remove=["StubSession"],
+        )
+
+    @staticmethod
+    def _classify_fork(current_type: str | None, has_parent: bool) -> LabelTransition:
+        if current_type == "ForkedSession":
+            return LabelTransition()
+        if current_type in ("RootSession", "SubSession"):
+            return LabelTransition(
+                add=["ForkedSession", "SST_EVENT"],
+                remove=[current_type, "StubSession"],
+            )
+        # bare session (current_type is None)
+        return LabelTransition(
+            add=["Session", "ForkedSession", "SST_EVENT"],
+            remove=["StubSession"],
+        )
+
 
 class SessionHandler:
     """Handles session lifecycle events.

context_intelligence_server/main.py

@@ -1140,6 +1140,11 @@ async def post_events(
         body = await http_request.body()
         body_obj = json.loads(body)
         body_obj["created_by"] = contributor_id  # overwrite, never setdefault
+        # Lift the optional top-level working_dir envelope field into data so the
+        # Session-node write sees it; absent/empty leaves Session.working_dir null
+        # for a later event to populate.
+        if request.working_dir and isinstance(body_obj.get("data"), dict):
+            body_obj["data"]["working_dir"] = request.working_dir
         body = json.dumps(body_obj, separators=(",", ":")).encode()
         await registry.queue_manager.append(worker_key, body)
         # Bytes are on disk: the key may be burned now (a failed append

context_intelligence_server/models.py

@@ -14,10 +14,19 @@ class EventRequest(BaseModel):
     The Amplifier client must always supply workspace on every event.
     Events without workspace (e.g. an incorrectly configured hook) are
     rejected at the endpoint with HTTP 422.
+
+    working_dir is OPTIONAL — the bundle hook emits it
+    as a top-level envelope field alongside workspace, but older clients/events
+    won't have it. Absent/empty is fine and leaves the Session node's
+    working_dir property null. Populate-if-missing: the Session node's
+    working_dir is filled in by the first subsequent event (including a
+    re-import via the upload CLI) that carries a non-empty value, but an
+    already-populated value is never overwritten.
     """
 
     event: str
     workspace: str
+    working_dir: str | None = None
     idempotency_key: str | None = None
     data: dict[str, Any]
 
@@ -29,6 +38,20 @@ def workspace_must_not_be_empty(cls, v: str) -> str:
             raise ValueError("workspace must not be empty")
         return v
 
+    @field_validator("working_dir")
+    @classmethod
+    def working_dir_must_not_be_blank(cls, v: str | None) -> str | None:
+        """Allow ``None`` (working_dir is optional, unlike workspace) but reject
+        blank/whitespace-only strings.
+
+        Mirrors ``workspace_must_not_be_empty``'s normalize-or-reject stance: a
+        whitespace-only value (e.g. ``"   "``) is never a legitimate path and must
+        not write through to the Session node verbatim.
+        """
+        if v is not None and not v.strip():
+            raise ValueError("working_dir must not be blank")
+        return v
+
 
 class EventResponse(BaseModel):
     """Response returned after an event is accepted."""

context_intelligence_server/neo4j_store.py

@@ -1009,12 +1009,19 @@ def _chunk_list(
 def _build_node_props(data: dict[str, Any], workspace: str) -> dict[str, Any]:
     """Assemble the sanitized props dict for a single node row.
 
-    ``labels`` and ``created_by`` are excluded from the returned dict:
+    ``labels``, ``created_by``, and ``working_dir`` are excluded from the returned dict:
     - ``labels`` are applied separately via ``SET n:Label`` statements (not stored as props).
     - ``created_by`` travels ONLY as the ``$created_by`` query param (never a node property)
       so it cannot affect node identity or clobber the ``ON CREATE SET n.created_by`` stamp.
+    - ``working_dir`` is excluded from the blind ``SET n += row.props`` so the
+      Session-node write can apply ``coalesce(n.working_dir, row.working_dir)``
+      instead of last-write-wins -- an already-set working_dir is never clobbered.
     """
-    raw = {k: v for k, v in data.items() if k not in ("labels", "created_by")}
+    raw = {
+        k: v
+        for k, v in data.items()
+        if k not in ("labels", "created_by", "working_dir")
+    }
     _convert_temporal_props(raw)  # ISO str -> datetime, in place
     props = Neo4jGraphStore._sanitize_properties(raw)
     props["workspace"] = workspace
@@ -1056,6 +1063,13 @@ async def _write_batch(
         row: dict[str, Any] = {"node_id": node_id, "props": props}
 
         if "Session" in labels:
+            # working_dir is a Session-only property carried as a separate
+            # top-level row key (not in props) so the MERGE below can coalesce
+            # it rather than blindly overwrite. Omitted when empty/absent, so
+            # coalesce(n.working_dir, null) is a no-op for such rows.
+            working_dir_value = data.get("working_dir")
+            if working_dir_value:
+                row["working_dir"] = working_dir_value
             session_rows.append(row)
         else:
             other_rows.append(row)
@@ -1085,7 +1099,12 @@ async def _write_batch(
             f"MERGE (n:{_UNIVERSAL_NODE_LABEL} "
             "{node_id: row.node_id, workspace: row.props.workspace}) "
             "ON CREATE SET n.created_by = $created_by "
-            "SET n += row.props, n:Session",
+            "SET n += row.props, n:Session "
+            # Populate-if-missing: fill working_dir only when it is still null,
+            # never clobber an already-set value (a concurrent/replica writer or
+            # an earlier event may have populated it). Rows without a working_dir
+            # carry a null row key, so this is a no-op for them.
+            "SET n.working_dir = coalesce(n.working_dir, row.working_dir)",
             rows=session_rows,
             created_by=created_by,
         )

context_intelligence_server/services.py

@@ -343,10 +343,28 @@ async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> No
         if existing is not None:
             # Upsert a stub so this worker's own flush uses MERGE (idempotent)
             # instead of racing a second worker into creating a duplicate node.
-            await self.graph.upsert_node(
-                session_id,
-                {"labels": ["Session"], "status": "running", "session_id": session_id},
-            )
+            stub_data: dict[str, Any] = {
+                "labels": ["Session"],
+                "status": "running",
+                "session_id": session_id,
+            }
+            # Populate-if-missing: backfill working_dir on a node created before
+            # working_dir was known (or by a bare reference). Only when the event
+            # supplies one AND the node still lacks it; the DB-level coalesce
+            # 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"]
+            # 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
 
@@ -366,6 +384,10 @@ async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> No
             node_data["started_at"] = _ts
         if "agent" in data:
             node_data["agent"] = data["agent"]
+        # Lift working_dir onto the Session node when the event carries one;
+        # absent/empty leaves it null for a later event to populate.
+        if data.get("working_dir"):
+            node_data["working_dir"] = data["working_dir"]
 
         await self.graph.upsert_node(session_id, node_data)
         self._seen_sessions.add(session_id)  # only cache after successful write

pyproject.toml

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