diff --git a/context_intelligence_server/models.py b/context_intelligence_server/models.py
index 03cf0793..ccdb1832 100644
--- a/context_intelligence_server/models.py
+++ b/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."""
diff --git a/context_intelligence_server/neo4j_store.py b/context_intelligence_server/neo4j_store.py
index 821db6c5..fbfd1599 100644
--- a/context_intelligence_server/neo4j_store.py
+++ b/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,
)
diff --git a/context_intelligence_server/pipeline.py b/context_intelligence_server/pipeline.py
index 7310fd9c..99467e6b 100644
--- a/context_intelligence_server/pipeline.py
+++ b/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 = (
diff --git a/context_intelligence_server/registry.py b/context_intelligence_server/registry.py
index d31143d2..22ec9536 100644
--- a/context_intelligence_server/registry.py
+++ b/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:
diff --git a/context_intelligence_server/services.py b/context_intelligence_server/services.py
index b5f0230b..6c344b9a 100644
--- a/context_intelligence_server/services.py
+++ b/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
diff --git a/docs/architecture/03-graph-model.dot b/docs/architecture/03-graph-model.dot
index 0efb28a8..273684b7 100644
--- a/docs/architecture/03-graph-model.dot
+++ b/docs/architecture/03-graph-model.dot
@@ -41,6 +41,7 @@ digraph graph_model {
| :RootSession :Session :SST_EVENT |
| node_id: session UUID |
| workspace: string |
+ | working_dir: path (populate-if-missing) |
| started_at / ended_at: ISO 8601 |
| status: "running" → closed |
>
@@ -52,6 +53,7 @@ digraph graph_model {
| :SubSession :Session :SST_EVENT |
| node_id: child session UUID |
| workspace: string |
+ | working_dir: path (populate-if-missing) |
| started_at / ended_at: ISO 8601 |
>
]
@@ -62,6 +64,7 @@ digraph graph_model {
| :ForkedSession :Session :SST_EVENT |
| node_id: child session UUID |
| workspace: may be null |
+ | working_dir: path (populate-if-missing) |
| started_at: ISO 8601 |
| Fork guard: permanent classification |
>
diff --git a/docs/architecture/03-graph-model.png b/docs/architecture/03-graph-model.png
index 061f8d75..c6655a7f 100644
Binary files a/docs/architecture/03-graph-model.png and b/docs/architecture/03-graph-model.png differ
diff --git a/tests/handlers/data_layer_1/test_default.py b/tests/handlers/data_layer_1/test_default.py
index fa22c9e7..f0977058 100644
--- a/tests/handlers/data_layer_1/test_default.py
+++ b/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'."""
diff --git a/tests/integration/test_crash_recovery.py b/tests/integration/test_crash_recovery.py
index 61fb544a..94e1895d 100644
--- a/tests/integration/test_crash_recovery.py
+++ b/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()
diff --git a/tests/integration/test_working_dir_attribution.py b/tests/integration/test_working_dir_attribution.py
new file mode 100644
index 00000000..eff573a7
--- /dev/null
+++ b/tests/integration/test_working_dir_attribution.py
@@ -0,0 +1,175 @@
+"""Integration — working-dir attribution survives the durable queue.
+
+The folder a session ran in arrives as a TOP-LEVEL envelope field on POST
+/events. ``post_events`` persists the raw request body verbatim, so the value
+is on disk with the event; the drainer reads it back off the queue line and
+attributes the Session node to it.
+
+That indirection is the whole point, and these tests are its regression guard:
+a worker respawned by crash recovery or dead-letter replay never sees the
+original HTTP request, so any design that binds working_dir to the in-memory
+worker loses it on every restart. Draining a persisted line with a FRESH
+worker — exactly what recovery does — is what proves it does not.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import httpx
+
+import context_intelligence_server.main as main_module
+from context_intelligence_server import registry as registry_module
+from context_intelligence_server.queue_manager import QueueManager
+from context_intelligence_server.registry import SessionRegistry, SessionWorker
+from context_intelligence_server.services import HookStateService
+
+WORKING_DIR = "/home/user/project"
+
+
+def _line(event: str, session_id: str, working_dir: str | None) -> bytes:
+ obj: dict[str, object] = {
+ "event": event,
+ "workspace": "-home-user-project",
+ "data": {"session_id": session_id, "timestamp": "2024-01-01T00:00:00+00:00"},
+ }
+ if working_dir is not None:
+ obj["working_dir"] = working_dir
+ return json.dumps(obj).encode("utf-8")
+
+
+async def _drain_once(registry: SessionRegistry, worker: SessionWorker) -> None:
+ """Run the real drain loop until the worker's queue is empty."""
+ task = asyncio.create_task(registry.drain_worker(worker, flush_timeout=10.0))
+ for _ in range(400):
+ await asyncio.sleep(0.01)
+ if (await registry.queue_manager.read_batch(worker.session_id, 10)).lines == []:
+ break
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
+
+
+async def test_post_events_persists_working_dir_on_the_queue_line(
+ client: httpx.AsyncClient,
+) -> None:
+ """The durable record carries working_dir at the top level, beside workspace."""
+ resp = await client.post(
+ "/events",
+ json={
+ "event": "session:start",
+ "workspace": "-home-user-project",
+ "working_dir": WORKING_DIR,
+ "data": {
+ "session_id": "sess-wd-persist",
+ "timestamp": "2024-01-01T00:00:00+00:00",
+ },
+ },
+ )
+ assert resp.status_code == 202
+
+ batch = await main_module.registry.queue_manager.read_batch("sess-wd-persist", 10)
+ assert len(batch.lines) == 1
+ obj = json.loads(batch.lines[0].decode("utf-8"))
+ assert obj["working_dir"] == WORKING_DIR
+ # Session attribute, not event content: it must NOT be smuggled into data,
+ # which is stored verbatim as a blob on every Event node.
+ assert "working_dir" not in obj["data"]
+
+
+async def test_working_dir_reaches_the_session_node_after_a_restart() -> None:
+ """A worker built WITHOUT any HTTP context still attributes the session.
+
+ This is the crash-recovery shape: recovery respawns a drainer from the
+ queue alone (main._recover_one_session), so the only surviving copy of the
+ working directory is the one on the persisted line.
+ """
+ settings = registry_module.get_settings()
+ sid = "sess-wd-recovered"
+ qm = QueueManager(queues_dir=Path(settings.queues_path))
+ await qm.append(sid, _line("session:start", sid, WORKING_DIR))
+
+ registry = SessionRegistry()
+ worker = SessionWorker(
+ session_id=sid,
+ workspace="-home-user-project",
+ services=HookStateService(workspace="-home-user-project"),
+ )
+ worker.services.graph.flush = AsyncMock() # type: ignore[method-assign]
+ worker.services.graph.close = AsyncMock() # type: ignore[method-assign]
+ registry._register_for_test(worker)
+
+ await _drain_once(registry, worker)
+
+ node = await worker.services.graph.get_node(sid)
+ assert node is not None
+ assert node.get("working_dir") == WORKING_DIR
+
+
+async def test_reimport_backfills_a_session_node_that_predates_working_dir() -> None:
+ """Re-ingesting an old session fills in a Session node that lacks the folder.
+
+ A node written before working_dir was recorded takes the node-exists branch
+ of ensure_session_node on the next drain. Populate-if-missing is what makes
+ the already-ingested corpus recoverable rather than permanently unattributed.
+ """
+ settings = registry_module.get_settings()
+ sid = "sess-wd-backfill"
+ qm = QueueManager(queues_dir=Path(settings.queues_path))
+ await qm.append(sid, _line("session:resume", sid, WORKING_DIR))
+
+ registry = SessionRegistry()
+ services = HookStateService(workspace="-home-user-project")
+ # A node from an earlier run: no working_dir, and NOT in _seen_sessions.
+ await services.graph.upsert_node(
+ sid, {"labels": ["Session"], "status": "running", "session_id": sid}
+ )
+ services.graph.flush = AsyncMock() # type: ignore[method-assign]
+ services.graph.close = AsyncMock() # type: ignore[method-assign]
+ worker = SessionWorker(
+ session_id=sid, workspace="-home-user-project", services=services
+ )
+ registry._register_for_test(worker)
+
+ await _drain_once(registry, worker)
+
+ node = await services.graph.get_node(sid)
+ assert node is not None
+ assert node.get("working_dir") == WORKING_DIR
+
+
+async def test_existing_working_dir_is_never_re_attributed() -> None:
+ """An event reporting a different folder does not move an attributed session."""
+ settings = registry_module.get_settings()
+ sid = "sess-wd-stable"
+ qm = QueueManager(queues_dir=Path(settings.queues_path))
+ await qm.append(sid, _line("session:resume", sid, "/somewhere/else"))
+
+ registry = SessionRegistry()
+ services = HookStateService(workspace="-home-user-project")
+ await services.graph.upsert_node(
+ sid,
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": sid,
+ "working_dir": WORKING_DIR,
+ },
+ )
+ services.graph.flush = AsyncMock() # type: ignore[method-assign]
+ services.graph.close = AsyncMock() # type: ignore[method-assign]
+ worker = SessionWorker(
+ session_id=sid, workspace="-home-user-project", services=services
+ )
+ registry._register_for_test(worker)
+
+ await _drain_once(registry, worker)
+
+ node = await services.graph.get_node(sid)
+ assert node is not None
+ assert node.get("working_dir") == WORKING_DIR
diff --git a/tests/neo4j/test_working_dir_coalesce.py b/tests/neo4j/test_working_dir_coalesce.py
new file mode 100644
index 00000000..a49a5d37
--- /dev/null
+++ b/tests/neo4j/test_working_dir_coalesce.py
@@ -0,0 +1,184 @@
+"""Real-Neo4j behavioral gates for working_dir populate-if-missing.
+
+``ensure_session_node`` guards against overwriting an already-attributed
+session in Python, but that guard only covers writers that went through this
+process's in-memory node cache. Two workers draining concurrently, a replayed
+batch, or a second server instance all reach the MERGE directly — so the
+"never re-attribute" rule is enforced a second time in Cypher:
+
+ SET n.working_dir = coalesce(n.working_dir, row.working_dir)
+
+These gates drive the REAL flush path against a REAL Neo4j and verify the
+resulting property with a raw sync driver. A unit test can only prove the
+Cypher string contains ``coalesce``; only this proves Neo4j honours it.
+
+ISOLATION GUARANTEE
+--------------------
+Uses ONLY the ephemeral Docker container from tests/neo4j/conftest.py
+(random ports, remove=True, fixture-injected credentials). No production
+Neo4j endpoint is referenced anywhere in this module.
+
+Gates
+------
+Gate A — first write lands: a Session node is created carrying working_dir.
+Gate B — second write with a DIFFERENT value does not overwrite it.
+Gate C — a row with NO working_dir does not null out an existing value.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from neo4j import GraphDatabase
+
+from context_intelligence_server.neo4j_store import Neo4jGraphStore
+
+pytestmark = pytest.mark.neo4j
+
+
+def _sync_driver(container: dict[str, Any]): # type: ignore[return]
+ """Return a synchronous Neo4j driver for verification/teardown queries."""
+ return GraphDatabase.driver(
+ container["bolt_url"],
+ auth=(container["user"], container["password"]),
+ )
+
+
+def _query_working_dir(
+ container: dict[str, Any], node_id: str, workspace: str
+) -> str | None:
+ """Return ``n.working_dir`` for (node_id, workspace), or None if unset."""
+ driver = _sync_driver(container)
+ try:
+ with driver.session() as session:
+ rec = session.run(
+ "MATCH (n {node_id: $nid, workspace: $ws}) RETURN n.working_dir AS wd",
+ nid=node_id,
+ ws=workspace,
+ ).single()
+ return rec["wd"] if rec else None
+ finally:
+ driver.close()
+
+
+def _cleanup_workspace(container: dict[str, Any], workspace: str) -> None:
+ """Delete all nodes (and their relationships) in *workspace*."""
+ driver = _sync_driver(container)
+ try:
+ with driver.session() as session:
+ session.run("MATCH (n {workspace: $ws}) DETACH DELETE n", ws=workspace)
+ finally:
+ driver.close()
+
+
+def _store(container: dict[str, Any], workspace: str) -> Neo4jGraphStore:
+ return Neo4jGraphStore(
+ uri=container["bolt_url"],
+ auth=(container["user"], container["password"]),
+ workspace=workspace,
+ )
+
+
+@pytest.mark.neo4j
+class TestWorkingDirCoalesceGates:
+ """working_dir is written once and never re-written, enforced by Neo4j."""
+
+ async def test_first_write_lands_on_the_session_node(
+ self, neo4j_container: dict[str, Any]
+ ) -> None:
+ ws = "wd-gate-a"
+ sid = "wd-a-session"
+ store = _store(neo4j_container, ws)
+ try:
+ await store.upsert_node(
+ sid,
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": sid,
+ "working_dir": "/home/user/project",
+ },
+ )
+ await store.flush()
+ assert _query_working_dir(neo4j_container, sid, ws) == "/home/user/project"
+ finally:
+ await store.close()
+ _cleanup_workspace(neo4j_container, ws)
+
+ async def test_second_write_does_not_overwrite(
+ self, neo4j_container: dict[str, Any]
+ ) -> None:
+ """A later flush carrying a DIFFERENT working_dir is ignored.
+
+ This is the concurrency case the Python guard cannot cover: a second
+ writer never consulted this process's cache, so only coalesce keeps the
+ session attributed to the folder it actually ran in.
+ """
+ ws = "wd-gate-b"
+ sid = "wd-b-session"
+ first = _store(neo4j_container, ws)
+ second = _store(neo4j_container, ws)
+ try:
+ await first.upsert_node(
+ sid,
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": sid,
+ "working_dir": "/original/path",
+ },
+ )
+ await first.flush()
+
+ await second.upsert_node(
+ sid,
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": sid,
+ "working_dir": "/different/path",
+ },
+ )
+ await second.flush()
+
+ assert _query_working_dir(neo4j_container, sid, ws) == "/original/path"
+ finally:
+ await first.close()
+ await second.close()
+ _cleanup_workspace(neo4j_container, ws)
+
+ async def test_row_without_working_dir_does_not_null_existing(
+ self, neo4j_container: dict[str, Any]
+ ) -> None:
+ """Rows carrying no working_dir leave an attributed session untouched.
+
+ Most Session writes (status flips, label enrichment, touch_session)
+ carry no working_dir at all. Those must be no-ops for the property,
+ not a coalesce against null that erases it.
+ """
+ ws = "wd-gate-c"
+ sid = "wd-c-session"
+ store = _store(neo4j_container, ws)
+ try:
+ await store.upsert_node(
+ sid,
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": sid,
+ "working_dir": "/home/user/project",
+ },
+ )
+ await store.flush()
+
+ # A later enrichment write with no working_dir key at all.
+ await store.upsert_node(
+ sid, {"labels": ["Session", "RootSession"], "status": "closed"}
+ )
+ await store.flush()
+
+ assert _query_working_dir(neo4j_container, sid, ws) == "/home/user/project"
+ finally:
+ await store.close()
+ _cleanup_workspace(neo4j_container, ws)
diff --git a/tests/test_drain_lifecycle_logging.py b/tests/test_drain_lifecycle_logging.py
index 92b78319..fcaa848a 100644
--- a/tests/test_drain_lifecycle_logging.py
+++ b/tests/test_drain_lifecycle_logging.py
@@ -80,7 +80,11 @@ async def test_g1a_cancelled_during_dispatch_logs_info_site_dispatch(
release = asyncio.Event()
async def _blocking_process(
- worker: object, event: str, data: object, handlers: object
+ worker: object,
+ event: str,
+ data: object,
+ handlers: object,
+ **_kw: object,
) -> None:
started.set()
await release.wait() # never set -- cancellation always wins here
diff --git a/tests/test_drain_supervision.py b/tests/test_drain_supervision.py
index 3c439972..d83f8879 100644
--- a/tests/test_drain_supervision.py
+++ b/tests/test_drain_supervision.py
@@ -117,7 +117,11 @@ async def close(self) -> None:
async def _accumulate(
- worker: SessionWorker, event: str, data: object, handlers: object
+ worker: SessionWorker,
+ event: str,
+ data: object,
+ handlers: object,
+ **_kw: object,
) -> None:
"""Stand-in for ``process_event``: buffers the event name on the fake
graph, exactly like ``_FaultInjectableGraph``'s harness in the sibling
@@ -698,7 +702,7 @@ async def test_committed_offset_freezes_AT_the_terminal_line(self) -> None:
mock_finalize.assert_awaited_once()
pending = await qm.read_batch(sid, 10)
assert len(pending.records) == 1
- event, _ws, _data = reg._parse_line(pending.records[0].raw)
+ event, _ws, _wd, _data = reg._parse_line(pending.records[0].raw)
assert event == "session:end"
async def test_recover_reports_a_terminal_but_unfinalized_session(self) -> None:
diff --git a/tests/test_durable_append_framing.py b/tests/test_durable_append_framing.py
index 18040c2a..dc2dbfd9 100644
--- a/tests/test_durable_append_framing.py
+++ b/tests/test_durable_append_framing.py
@@ -940,7 +940,7 @@ def test_captured_valid_large_event_parses_cleanly() -> None:
assert len(raw) > 1024 * 1024
assert raw.count(b"\n") == 0
- event, workspace, data = SessionRegistry._parse_line(raw)
+ event, workspace, _working_dir, data = SessionRegistry._parse_line(raw)
assert event
assert workspace
assert isinstance(data, dict)
diff --git a/tests/test_finalize_delete_ordering.py b/tests/test_finalize_delete_ordering.py
index 3815a3ad..c5888503 100644
--- a/tests/test_finalize_delete_ordering.py
+++ b/tests/test_finalize_delete_ordering.py
@@ -70,7 +70,11 @@ async def close(self) -> None:
async def _accumulate(
- worker: SessionWorker, event: str, data: object, handlers: object
+ worker: SessionWorker,
+ event: str,
+ data: object,
+ handlers: object,
+ **_kw: object,
) -> None:
"""Stand-in for ``process_event``: buffers the event name on the fake graph."""
worker.services.graph.buffer.add(event)
diff --git a/tests/test_large_event_tail_drop.py b/tests/test_large_event_tail_drop.py
index 3c89e2c1..bac397a3 100644
--- a/tests/test_large_event_tail_drop.py
+++ b/tests/test_large_event_tail_drop.py
@@ -133,7 +133,9 @@ async def test_prefix_and_tail_persist_oversized_dead_lettered(self) -> None:
worker.services.graph = fake # type: ignore[assignment]
reg._register_for_test(worker)
- async def _process(w: object, event: str, data: object, h: object) -> None:
+ async def _process(
+ w: object, event: str, data: object, h: object, **_kw: object
+ ) -> None:
fake.buffer.add(event)
with patch(
diff --git a/tests/test_models.py b/tests/test_models.py
index 1cad9da8..944abc78 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -74,6 +74,43 @@ def test_event_request_workspace_non_empty_accepted():
assert req.workspace == "my-project-slug"
+def test_event_request_working_dir_defaults_none():
+ """working_dir defaults to None — "not reported", NOT the empty path."""
+ req = EventRequest(
+ event="tool:pre",
+ workspace="main",
+ data={"session_id": "abc123"},
+ )
+ assert req.working_dir is None
+
+
+def test_event_request_accepts_working_dir():
+ """working_dir is parsed off the top-level envelope, not from data."""
+ req = EventRequest(
+ event="tool:pre",
+ workspace="main",
+ working_dir="/home/user/project",
+ data={"session_id": "abc123"},
+ )
+ assert req.working_dir == "/home/user/project"
+
+
+def test_event_request_rejects_blank_working_dir():
+ """A whitespace-only working_dir is never a legitimate path — reject it.
+
+ Absent is fine (None); blank is a malformed client, and must not reach the
+ Session node verbatim where it would satisfy the populate-if-missing guard
+ and permanently block the real value.
+ """
+ with pytest.raises(ValidationError):
+ EventRequest(
+ event="tool:pre",
+ workspace="main",
+ working_dir=" ",
+ data={"session_id": "abc123"},
+ )
+
+
def test_event_request_data_without_session_id():
"""EventRequest accepts data dict that has no session_id key."""
req = EventRequest(
diff --git a/tests/test_neo4j_store.py b/tests/test_neo4j_store.py
index 2051e04f..2f919e43 100644
--- a/tests/test_neo4j_store.py
+++ b/tests/test_neo4j_store.py
@@ -3363,3 +3363,104 @@ async def test_run_repair_still_fails_closed_on_genuine_conflict(self) -> None:
with pytest.raises(RuntimeError, match="doctor --fix"):
await run_repair(_FakeDriver(session))
+
+
+# ---------------------------------------------------------------------------
+# working_dir: populate-if-missing at the Session MERGE
+# ---------------------------------------------------------------------------
+
+
+class TestWorkingDirCoalesce:
+ """working_dir is coalesced, never blind-overwritten, on Session nodes.
+
+ Holding it out of ``row.props`` and applying
+ ``coalesce(n.working_dir, row.working_dir)`` makes "an already-attributed
+ session is never re-attributed" a DATABASE guarantee, so it also holds for
+ a concurrent writer or a replayed batch — not just for the in-process
+ guard in ensure_session_node.
+ """
+
+ def test_build_node_props_excludes_working_dir(self) -> None:
+ from context_intelligence_server.neo4j_store import _build_node_props
+
+ props = _build_node_props(
+ {"labels": ["Session"], "status": "running", "working_dir": "/p"},
+ "ws-1",
+ )
+ assert "working_dir" not in props, (
+ "working_dir must NOT ride in row.props — the blind `SET n += row.props` "
+ "would overwrite an already-set value, defeating populate-if-missing"
+ )
+ assert props["status"] == "running"
+
+ def test_write_batch_cypher_coalesces_working_dir(self) -> None:
+ import inspect
+
+ from context_intelligence_server import neo4j_store
+
+ source = inspect.getsource(neo4j_store._write_batch)
+ assert "coalesce(n.working_dir, row.working_dir)" in source, (
+ "the Session MERGE must coalesce working_dir rather than overwrite it"
+ )
+
+ async def test_write_batch_passes_working_dir_as_row_key(self) -> None:
+ """The Session row carries working_dir as a top-level key, not in props."""
+ from context_intelligence_server.neo4j_store import _write_batch
+
+ captured: list[dict[str, object]] = []
+
+ class _Tx:
+ async def run(self, statement: str, **kwargs: object) -> object:
+ captured.append({"statement": statement, "kwargs": kwargs})
+
+ class _R:
+ async def consume(self) -> None:
+ return None
+
+ return _R()
+
+ await _write_batch(
+ _Tx(),
+ {
+ "s1": {
+ "labels": ["Session"],
+ "status": "running",
+ "working_dir": "/home/user/project",
+ }
+ },
+ {},
+ [],
+ "ws-1",
+ )
+ session_calls = [c for c in captured if "n:Session" in str(c["statement"])]
+ assert session_calls, "expected a Session MERGE statement"
+ rows = session_calls[0]["kwargs"]["rows"] # type: ignore[index]
+ assert rows[0]["working_dir"] == "/home/user/project"
+ assert "working_dir" not in rows[0]["props"]
+
+ async def test_write_batch_omits_working_dir_row_key_when_absent(self) -> None:
+ """No working_dir on the node => no row key => the coalesce is a no-op."""
+ from context_intelligence_server.neo4j_store import _write_batch
+
+ captured: list[dict[str, object]] = []
+
+ class _Tx:
+ async def run(self, statement: str, **kwargs: object) -> object:
+ captured.append({"statement": statement, "kwargs": kwargs})
+
+ class _R:
+ async def consume(self) -> None:
+ return None
+
+ return _R()
+
+ await _write_batch(
+ _Tx(),
+ {"s1": {"labels": ["Session"], "status": "running"}},
+ {},
+ [],
+ "ws-1",
+ )
+ session_calls = [c for c in captured if "n:Session" in str(c["statement"])]
+ rows = session_calls[0]["kwargs"]["rows"] # type: ignore[index]
+ assert "working_dir" not in rows[0]
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index dfe3bffb..69e6f57a 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -350,7 +350,35 @@ async def test_process_event_calls_ensure_session_node(
data = {"session_id": "sess-123"}
await process_event(mock_worker, "session:start", data, pipeline_handlers)
- mock_worker.services.ensure_session_node.assert_called_once_with("sess-123", data)
+ mock_worker.services.ensure_session_node.assert_called_once_with(
+ "sess-123", data, working_dir=None
+ )
+
+
+async def test_process_event_forwards_working_dir_to_ensure_session_node(
+ mock_worker: MagicMock,
+ pipeline_handlers: Any,
+) -> None:
+ """The envelope working_dir reaches the Session node write.
+
+ It is passed ALONGSIDE data rather than injected into it: working_dir is a
+ session attribute, and data is stored verbatim as a blob on every Event
+ node, so smuggling it there would duplicate it per event.
+ """
+ from context_intelligence_server.pipeline import process_event
+
+ data = {"session_id": "sess-123"}
+ await process_event(
+ mock_worker,
+ "session:start",
+ data,
+ pipeline_handlers,
+ working_dir="/home/user/project",
+ )
+ mock_worker.services.ensure_session_node.assert_called_once_with(
+ "sess-123", data, working_dir="/home/user/project"
+ )
+ assert "working_dir" not in data
async def test_process_event_missing_session_id_skips_ensure_but_dispatches(
diff --git a/tests/test_registry.py b/tests/test_registry.py
index 901e00e7..6be2bb46 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -761,7 +761,7 @@ async def test_process_one_exception_logged(
worker.services.graph.flush = AsyncMock() # type: ignore[method-assign]
reg._register_for_test(worker)
- async def mock_process(w, event, data, handlers):
+ async def mock_process(w, event, data, handlers, **_kw):
raise ValueError("handler exploded")
with (
@@ -1016,7 +1016,7 @@ async def test_error_count_incremented_on_process_event_failure(
call_count = 0
async def mock_process(
- w: object, event: str, data: object, handlers: object
+ w: object, event: str, data: object, handlers: object, **_kw: object
) -> None:
nonlocal call_count
call_count += 1
@@ -1044,7 +1044,9 @@ async def test_session_end_drains_tail_to_eof(
sid = "s-tail"
processed: list[str] = []
- async def _capture(w: object, event: str, data: object, h: object) -> None:
+ async def _capture(
+ w: object, event: str, data: object, h: object, **_kw: object
+ ) -> None:
processed.append(event)
worker = SessionWorker(
@@ -1204,7 +1206,9 @@ async def test_only_the_poison_line_is_dead_lettered_clean_buffer(
)
worker.services.graph = fake # type: ignore[assignment]
- async def _process(w: object, event: str, data: object, h: object) -> None:
+ async def _process(
+ w: object, event: str, data: object, h: object, **_kw: object
+ ) -> None:
# process_event buffers the line's write (here: the event name).
fake.buffer.add(event)
@@ -1709,7 +1713,9 @@ async def test_exhausted_batch_per_line_success_increments_written(
)
worker.services.graph = fake # type: ignore[assignment]
- async def _process(w: object, event: str, data: object, h: object) -> None:
+ async def _process(
+ w: object, event: str, data: object, h: object, **_kw: object
+ ) -> None:
fake.buffer.add(event)
reg._register_for_test(worker)
@@ -2186,3 +2192,56 @@ def test_invariant_violation_is_observed_and_not_overwritten(
# 3. Worker is returned — no exception raised
assert worker is not None
+
+
+class TestParseLineWorkingDir:
+ """_parse_line reads working_dir off the DURABLE queue line.
+
+ This is the load-bearing property of working-dir attribution: the value is
+ persisted with the event by post_events (which stores the raw request body
+ verbatim), so it is still there for a worker respawned by crash recovery or
+ dead-letter replay — neither of which ever sees the original HTTP request.
+ """
+
+ def test_parse_line_reads_top_level_working_dir(self) -> None:
+ from context_intelligence_server.registry import SessionRegistry
+
+ raw = json.dumps(
+ {
+ "event": "tool:pre",
+ "workspace": "-ws",
+ "working_dir": "/home/user/project",
+ "data": {"session_id": "s1"},
+ }
+ ).encode("utf-8")
+ event, workspace, working_dir, data = SessionRegistry._parse_line(raw)
+ assert event == "tool:pre"
+ assert workspace == "-ws"
+ assert working_dir == "/home/user/project"
+ assert data == {"session_id": "s1"}
+
+ def test_parse_line_working_dir_absent_is_none(self) -> None:
+ """A line with no working_dir yields None, not "" — absent != blank."""
+ from context_intelligence_server.registry import SessionRegistry
+
+ raw = json.dumps(
+ {"event": "tool:pre", "workspace": "-ws", "data": {"session_id": "s1"}}
+ ).encode("utf-8")
+ _event, _ws, working_dir, _data = SessionRegistry._parse_line(raw)
+ assert working_dir is None
+
+ def test_parse_line_working_dir_empty_string_is_none(self) -> None:
+ """An empty working_dir normalizes to None so it cannot satisfy the
+ populate-if-missing guard and block the real value later."""
+ from context_intelligence_server.registry import SessionRegistry
+
+ raw = json.dumps(
+ {
+ "event": "tool:pre",
+ "workspace": "-ws",
+ "working_dir": "",
+ "data": {"session_id": "s1"},
+ }
+ ).encode("utf-8")
+ _event, _ws, working_dir, _data = SessionRegistry._parse_line(raw)
+ assert working_dir is None
diff --git a/tests/test_services.py b/tests/test_services.py
index 7b37014b..93c6d87c 100644
--- a/tests/test_services.py
+++ b/tests/test_services.py
@@ -231,6 +231,73 @@ async def test_ensure_session_node_creates_root(self):
assert "RootSession" not in node["labels"]
assert node["status"] == "running"
+ async def test_ensure_session_node_lifts_working_dir(self):
+ """working_dir from the event envelope is stamped on a new Session node."""
+ svc = HookStateService()
+ await svc.ensure_session_node(
+ "session-wd",
+ {"started_at": "2024-01-01T00:00:00"},
+ working_dir="/home/user/project",
+ )
+ node = await svc.graph.get_node("session-wd")
+ assert node is not None
+ assert node["working_dir"] == "/home/user/project"
+
+ async def test_ensure_session_node_omits_absent_working_dir(self):
+ """No working_dir property is written when the event reports none."""
+ svc = HookStateService()
+ await svc.ensure_session_node(
+ "session-no-wd", {"started_at": "2024-01-01T00:00:00"}
+ )
+ node = await svc.graph.get_node("session-no-wd")
+ assert node is not None
+ assert "working_dir" not in node
+
+ async def test_ensure_session_node_backfills_existing_node(self):
+ """A Session node that already exists WITHOUT a working_dir gets one.
+
+ This is the re-import / crash-recovery path: the node survives from an
+ earlier run (or was stubbed by a delegation edge), so a fresh worker
+ takes the node-exists branch. Without populate-if-missing here, no
+ already-ingested session could ever be attributed to its folder.
+ """
+ svc = HookStateService()
+ # Simulate a node written by an earlier run, with no working_dir.
+ await svc.graph.upsert_node(
+ "session-backfill",
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": "session-backfill",
+ },
+ )
+ # A DIFFERENT worker (cold _seen_sessions) drains an event for it.
+ svc2 = HookStateService(graph_store=svc.graph)
+ await svc2.ensure_session_node(
+ "session-backfill", {}, working_dir="/home/user/project"
+ )
+ node = await svc.graph.get_node("session-backfill")
+ assert node is not None
+ assert node["working_dir"] == "/home/user/project"
+
+ async def test_ensure_session_node_never_overwrites_working_dir(self):
+ """An already-attributed session is not re-attributed by a later event."""
+ svc = HookStateService()
+ await svc.graph.upsert_node(
+ "session-set",
+ {
+ "labels": ["Session"],
+ "status": "running",
+ "session_id": "session-set",
+ "working_dir": "/original/path",
+ },
+ )
+ svc2 = HookStateService(graph_store=svc.graph)
+ await svc2.ensure_session_node("session-set", {}, working_dir="/different/path")
+ node = await svc.graph.get_node("session-set")
+ assert node is not None
+ assert node["working_dir"] == "/original/path"
+
async def test_ensure_session_node_is_idempotent(self):
"""ensure_session_node is a no-op when session_id was already processed."""
svc = HookStateService()