fix: attribute recorded sessions to the working directory they ran in

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 and re-imported archives may
+    omit it.  It is declared here for contract + validation only: the ingest
+    endpoint persists the RAW request body to the durable queue, so the field
+    reaches the drainer whether or not this model names it.  Absent/None leaves
+    the Session node's working_dir unset for a later event to populate;
+    populate-if-missing means an already-set 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,21 @@ 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.
+
+        ``None`` means "this client did not report a working directory" — which
+        is NOT the same as "the working directory is the empty string".  A
+        whitespace-only value (e.g. ``"   "``) is never a legitimate path and
+        must not reach 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

@@ -1035,12 +1035,20 @@ 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 held out of the blind ``SET n += row.props`` so the Session
+      MERGE can apply ``coalesce(n.working_dir, row.working_dir)`` instead of
+      last-write-wins — an already-attributed session is never re-attributed,
+      even by a concurrent writer or a replayed batch.
     """
-    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
@@ -1082,6 +1090,13 @@ async def _write_batch(
         row: dict[str, Any] = {"node_id": node_id, "props": props}
 
         if "Session" in labels:
+            # working_dir rides as a separate top-level row key (deliberately
+            # NOT inside row.props) so the MERGE below can coalesce it rather
+            # than blindly overwrite. Omitted when absent/empty, in which case
+            # row.working_dir is null and the coalesce is a no-op.
+            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)
@@ -1111,7 +1126,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 while it is still null,
+            # never clobber a value an earlier event (or a concurrent writer)
+            # already set. Rows without a working_dir carry a null row key, so
+            # this degrades to 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/pipeline.py

@@ -134,14 +134,22 @@ async def process_event(
     event: str,
     data: dict[str, Any],
     handlers: PipelineHandlers,
+    *,
+    working_dir: str | None = None,
 ) -> None:
     """Process one event through the always-default + enrichers pipeline.
 
+    *working_dir* is the envelope-level working directory for this event
+    (``None`` when the client did not report one).  It is a session attribute,
+    not event content, so it is passed alongside *data* rather than injected
+    into it — keeping it out of the ``data`` blob stored on every Event node.
+
     Steps
     -----
     1. Extract ``session_id`` from *data*.
     2. If *session_id* is present, call ``worker.services.ensure_session_node``
-       to idempotently create a Session node before any handler runs.
+       to idempotently create a Session node before any handler runs, passing
+       *working_dir* so the node is attributed to the folder it ran in.
     3. Blob processing: if session_id + timestamp + blob_store are all present,
        call ``process_event_data``.  Log a WARNING if blob_store is present but
        timestamp is missing.
@@ -166,7 +174,9 @@ async def process_event(
     try:
         # Step 2 — ensure Session node exists for known sessions
         if session_id:
-            await worker.services.ensure_session_node(session_id, data)
+            await worker.services.ensure_session_node(
+                session_id, data, working_dir=working_dir
+            )
 
         # Step 3 — blob processing (after ensure_session_node, before dispatch)
         timestamp: str | None = (

context_intelligence_server/registry.py

@@ -298,12 +298,20 @@ async def _process_one(
         event: str,
         data: dict[str, Any],
         handlers: Any,
+        *,
+        working_dir: str | None = None,
     ) -> None:
-        """Dispatch one event, update worker stats, and record to the ring buffer."""
+        """Dispatch one event, update worker stats, and record to the ring buffer.
+
+        *working_dir* is the envelope-level working directory read off the
+        queued line (``None`` when the line carries none); it is forwarded to
+        the pipeline so the Session node can be attributed to the folder the
+        session ran in.
+        """
         result = "ok"
         error = ""
         try:
-            await process_event(worker, event, data, handlers)
+            await process_event(worker, event, data, handlers, working_dir=working_dir)
             worker.last_event = event
             worker.last_event_time = time.time()
             worker.events_processed += 1
@@ -495,10 +503,28 @@ async def drain_worker(
                 return
 
     @staticmethod
-    def _parse_line(raw: bytes) -> tuple[str, str, dict[str, Any]]:
-        """Decode an appended event line (raw EventRequest JSON)."""
+    def _parse_line(raw: bytes) -> tuple[str, str, str | None, dict[str, Any]]:
+        """Decode an appended event line (raw EventRequest JSON).
+
+        ``working_dir`` is a TOP-LEVEL envelope field (a sibling of
+        ``workspace``), not part of ``data`` — the hook emits it as a session
+        attribute rather than event content.  Reading it here, off the durable
+        queue line, is what makes working-dir attribution crash-safe: the value
+        was persisted with the event by ``post_events`` (which stores the raw
+        request body verbatim), so a worker respawned by crash recovery or
+        dead-letter replay reads the same value from the same bytes.  Binding
+        it to the in-memory worker instead would lose it on every restart.
+
+        Returns ``None`` — never ``""`` — when the line carries no working_dir,
+        so "not reported" stays distinguishable from a blank path downstream.
+        """
         obj = json.loads(raw.decode("utf-8"))
-        return obj["event"], obj.get("workspace", ""), obj.get("data", {})
+        return (
+            obj["event"],
+            obj.get("workspace", ""),
+            obj.get("working_dir") or None,
+            obj.get("data", {}),
+        )
 
     async def _process_batch(
         self, worker: SessionWorker, batch: Batch, handlers: Any
@@ -529,8 +555,10 @@ async def _process_batch(
         terminal_at: int | None = None
         safe_count = 0
         for rec in batch.records:
-            event, _workspace, data = self._parse_line(rec.raw)
-            await self._process_one(worker, event, data, handlers)
+            event, _workspace, working_dir, data = self._parse_line(rec.raw)
+            await self._process_one(
+                worker, event, data, handlers, working_dir=working_dir
+            )
             if terminal_at is None:
                 if event in TERMINAL_EVENTS:
                     terminal_at = rec.start
@@ -569,7 +597,7 @@ async def _handle_exhausted_batch(
         worker.services.graph.discard_buffer()
         for rec in batch.records:
             try:
-                event, _ws, data = self._parse_line(rec.raw)
+                event, _ws, working_dir, data = self._parse_line(rec.raw)
             except Exception as exc:
                 # Unparseable: can't be a terminal record -- poison as before.
                 await qm.dead_letter(session_id, rec.raw, str(exc))  # no re-framing
@@ -591,7 +619,9 @@ async def _handle_exhausted_batch(
 
             wrote = False
             try:
-                await self._process_one(worker, event, data, handlers)
+                await self._process_one(
+                    worker, event, data, handlers, working_dir=working_dir
+                )
                 await self._flush_barrier(worker)
                 wrote = True
             except Exception as exc:

context_intelligence_server/services.py

@@ -247,9 +247,27 @@ def __init__(
     # Session node management
     # ------------------------------------------------------------------
 
-    async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> None:
+    async def ensure_session_node(
+        self,
+        session_id: str,
+        data: dict[str, Any],
+        *,
+        working_dir: str | None = None,
+    ) -> None:
         """Idempotently create a Session node in the graph for *session_id*.
 
+        *working_dir* is the folder the session ran in, read off the event
+        envelope.  It is applied POPULATE-IF-MISSING: written when the event
+        supplies one and the node does not already carry one, never overwritten
+        once set.  That rule is enforced twice — here (so an already-populated
+        node is left alone) and again at the Neo4j MERGE via ``coalesce``
+        (so a concurrent writer or a replayed batch cannot clobber it either).
+
+        Populate-if-missing is what makes backfill work: a Session node created
+        before working_dir was recorded — or created as a bare reference by a
+        delegation/fork edge — is filled in by the first later event that
+        carries one, including a re-import through the upload CLI.
+
         Uses a two-tier lookup for replay resilience:
 
         1. Fast path — if *session_id* is already in the in-memory
@@ -269,7 +287,14 @@ async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> No
         Only caches session_id after a successful write to ensure retry
         resilience on write failure.
         """
-        # Tier 1: fast path — warm cache hit
+        # Tier 1: fast path — warm cache hit.
+        # Safe with respect to working_dir: a worker only ever drains events for
+        # its OWN session (worker_key == session_id), and the hook stamps the
+        # same working_dir on every event of a session. So the FIRST call for
+        # this worker's own session already carries the value if it will ever
+        # carry one, and it is applied below before the cache is warmed. Nodes
+        # this worker stubs for OTHER sessions (a parent_id or a delegation's
+        # sub_session_id) are populated when that session's own worker runs.
         if session_id in self._seen_sessions:
             return
 
@@ -285,10 +310,19 @@ async def ensure_session_node(self, session_id: str, data: dict[str, Any]) -> No
             #   4. Worker B's flush later issues a fresh MERGE → duplicate node.
             # upsert_node uses union-merge for labels, so existing type labels
             # (e.g. "RootSession") are preserved — this call never strips labels.
-            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. This is the branch a re-import lands
+            # in (the node survives from the original ingest) and the branch a
+            # worker respawned by crash recovery lands in. Only write when this
+            # event supplies a working_dir AND the stored node still lacks one,
+            # so an already-attributed session is never re-attributed.
+            if working_dir and not existing.get("working_dir"):
+                stub_data["working_dir"] = working_dir
+            await self.graph.upsert_node(session_id, stub_data)
             self._seen_sessions.add(session_id)
             return
 
@@ -316,6 +350,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"]
+        # Attribute the session to the folder it ran in. Absent/None leaves the
+        # property unset so a later event (or a re-import) can populate it.
+        if working_dir:
+            node_data["working_dir"] = working_dir
 
         await self.graph.upsert_node(session_id, node_data)
         self._seen_sessions.add(session_id)  # only cache after successful write

docs/architecture/03-graph-model.dot

@@ -41,6 +41,7 @@ digraph graph_model {
                 <TR><TD><B>:RootSession :Session :SST_EVENT</B></TD></TR>
                 <TR><TD>node_id: session UUID</TD></TR>
                 <TR><TD>workspace: string</TD></TR>
+                <TR><TD>working_dir: path (populate-if-missing)</TD></TR>
                 <TR><TD>started_at / ended_at: ISO 8601</TD></TR>
                 <TR><TD>status: "running" &#8594; closed</TD></TR>
             </TABLE>>
@@ -52,6 +53,7 @@ digraph graph_model {
                 <TR><TD><B>:SubSession :Session :SST_EVENT</B></TD></TR>
                 <TR><TD>node_id: child session UUID</TD></TR>
                 <TR><TD>workspace: string</TD></TR>
+                <TR><TD>working_dir: path (populate-if-missing)</TD></TR>
                 <TR><TD>started_at / ended_at: ISO 8601</TD></TR>
             </TABLE>>
         ]
@@ -62,6 +64,7 @@ digraph graph_model {
                 <TR><TD><B>:ForkedSession :Session :SST_EVENT</B></TD></TR>
                 <TR><TD>node_id: child session UUID</TD></TR>
                 <TR><TD>workspace: may be null</TD></TR>
+                <TR><TD>working_dir: path (populate-if-missing)</TD></TR>
                 <TR><TD>started_at: ISO 8601</TD></TR>
                 <TR><TD><I>Fork guard: permanent classification</I></TD></TR>
             </TABLE>>

tests/handlers/data_layer_1/test_default.py

@@ -273,6 +273,29 @@ async def test_stores_data_property(self, services: HookStateService) -> None:
         assert data["custom_info"] == "extra-value"
 
 
+class TestDefaultHandlerWorkingDir:
+    """working_dir is a SESSION attribute — Event nodes must not duplicate it."""
+
+    async def test_event_node_does_not_carry_working_dir(
+        self, services: HookStateService
+    ) -> None:
+        """Event nodes carry no working_dir property.
+
+        The folder a session ran in is stored once on its Session node; every
+        Event is one HAS_EVENT hop away, so a per-event copy buys no query and
+        costs a string on the highest-volume write path in the system.
+        """
+        handler = DefaultHandler(services)
+        await handler(
+            "session:resume",
+            {"session_id": "s1", "timestamp": "2026-01-01T02:00:00Z"},
+        )
+        event_id = make_node_id("s1", "session:resume", "2026-01-01T02:00:00Z")
+        node = await services.graph.get_node(event_id)
+        assert node is not None
+        assert "working_dir" not in node
+
+
 class TestDefaultHandlerEdgeType:
     """HAS_EVENT edge has type='HAS_EVENT'."""
 

tests/integration/test_crash_recovery.py

@@ -32,7 +32,9 @@ async def test_no_loss_after_crash_mid_drain() -> None:
     reg1 = SessionRegistry()
     processed_first: list[str] = []
 
-    async def _proc1(w: object, event: str, data: object, h: object) -> None:
+    async def _proc1(
+        w: object, event: str, data: object, h: object, **_kw: object
+    ) -> None:
         processed_first.append(event)
         await asyncio.sleep(0.01)  # slow enough to be interrupted mid-drain
 
@@ -57,7 +59,9 @@ async def _proc1(w: object, event: str, data: object, h: object) -> None:
     reg2 = SessionRegistry()
     processed_second: list[str] = []
 
-    async def _proc2(w: object, event: str, data: object, h: object) -> None:
+    async def _proc2(
+        w: object, event: str, data: object, h: object, **_kw: object
+    ) -> None:
         processed_second.append(event)
 
     recovered = await reg2.queue_manager.recover()