Keep delegation pipeline-entries sibling-only

CHANGELOG.md

@@ -8,6 +8,17 @@ loosely while pre-1.0 (breaking changes can land on minor bumps).
 ## [Unreleased]
 
 ### Fixed
+- **Delegation pipeline memory is sibling-only.** The orchestrator's
+  `pipeline-entries` probe now lists rows pinned to the active conversation
+  (`conversation_id = ?`). Default `list_entries` still ORs in unscoped
+  (`NULL`) rows for agent `/mem` browse; that fallback was injecting HTTP
+  `/v1/memory` and CLI `/mem add` residue into sub-agent context as if
+  siblings had just written it.
+- **Sandbox FIFO no longer deadlocks a tenant.** `/write` and `/read` of a
+  named pipe used to block forever in `open()`. `/write` holds the
+  per-tenant sandbox mutex across that open, so a workspace `mkfifo`
+  stalled every same-tenant `/exec` and `/write` until restart. Both
+  paths now open `O_NONBLOCK` and reject non-regular files.
 - **A2A unary HTTP body cap.** `rpc_call` / `http_get` now refuse
   responses over 16 MiB (same ceiling as the inbound HTTP API and
   `kSseMaxEventBytes`). A remote that omitted `Content-Length` could

include/tenant_store.h

@@ -1179,6 +1179,14 @@ class TenantStore {
         // unscoped (`conversation_id IS NULL`); the OR-NULL fallback
         // keeps pre-migration entries visible everywhere.
         int64_t                  conversation_id       = 0;
+        // When true, a positive `conversation_id` matches only that
+        // conversation (`conversation_id = ?`).  The default OR-NULL
+        // fallback is for agent /mem browse and HTTP list, where
+        // unscoped / pre-migration rows should stay visible.  The
+        // delegation pipeline snapshot must opt out: HTTP `/v1/memory`
+        // and CLI `/mem add` rows with NULL conversation_id would
+        // otherwise be injected as "what siblings just wrote."
+        bool                     exact_conversation    = false;
         // Age-decay multiplier on BM25 scores.  When `age_now_epoch`
         // is non-zero, the search SQL multiplies each row's score by
         // a piecewise factor that drops from 1.0 at `valid_from = now`

src/api_server.cpp

@@ -6864,6 +6864,14 @@ StructuredMemoryReader make_structured_memory_reader_callback(
                 TenantStore::EntryFilter f;
                 f.limit = 15;
                 f.conversation_id = reader_conversation_id;
+                // Sibling snapshot only.  Default list_entries ORs in
+                // NULL conversation_id (HTTP admin / CLI unscoped
+                // rows, pre-migration residue).  Those are not sibling
+                // output; including them reintroduces the tenant-wide
+                // bleed this probe exists to prevent.  Must be in the
+                // SQL WHERE — post-filtering after LIMIT 15 can drop
+                // the actual sibling rows behind unscoped recency.
+                f.exact_conversation = true;
                 auto entries = tenants.list_entries(reader_tenant_id, f);
                 if (entries.empty()) return "(no entries)";
                 std::ostringstream out;

src/tenant_store.cpp

@@ -2593,11 +2593,16 @@ TenantStore::list_entries(int64_t tenant_id, const EntryFilter& f) const {
             sql += " AND e.valid_to IS NULL";
         }
         if (f.conversation_id > 0) {
-            // OR-NULL fallback: rows pinned to this conversation OR rows
-            // that are unscoped stay reachable.  Pre-migration entries
-            // (NULL conversation_id) are visible from every conversation.
-            sql += " AND (e.conversation_id = ? "
-                   "      OR e.conversation_id IS NULL)";
+            if (f.exact_conversation) {
+                sql += " AND e.conversation_id = ?";
+            } else {
+                // OR-NULL fallback: rows pinned to this conversation OR
+                // rows that are unscoped stay reachable.  Pre-migration
+                // entries (NULL conversation_id) are visible from every
+                // conversation.
+                sql += " AND (e.conversation_id = ? "
+                       "      OR e.conversation_id IS NULL)";
+            }
         }
         if (f.since > 0)             sql += " AND e.created_at >= ?";
         if (f.before_updated_at > 0) sql += " AND e.updated_at < ?";
@@ -2669,7 +2674,11 @@ TenantStore::list_entries(int64_t tenant_id, const EntryFilter& f) const {
         sql += " AND valid_to IS NULL";
     }
     if (f.conversation_id > 0) {
-        sql += " AND (conversation_id = ? OR conversation_id IS NULL)";
+        if (f.exact_conversation) {
+            sql += " AND conversation_id = ?";
+        } else {
+            sql += " AND (conversation_id = ? OR conversation_id IS NULL)";
+        }
     }
     if (f.since > 0)             sql += " AND created_at >= ?";
     if (f.before_updated_at > 0) sql += " AND updated_at < ?";

tests/test_memory_entries.cpp

@@ -719,6 +719,48 @@ TEST_CASE("conversation_id filter includes unscoped rows (OR-NULL fallback)") {
     }
 }
 
+TEST_CASE("exact_conversation excludes unscoped rows from pipeline snapshot") {
+    TempDb db;
+    TenantStore s;
+    s.open(db.path.string());
+    const int64_t tid = make_tenant(s, "pipeline-exact");
+    auto conv = s.create_conversation(tid, "delegating", "index", "");
+    auto other_conv = s.create_conversation(tid, "other", "index", "");
+
+    auto sibling = s.create_entry(tid, "project", "sibling scout notes",
+        "Fresh sibling output.", "", "[]", /*artifact=*/0, conv.id);
+    s.create_entry(tid, "project", "http admin import",
+        "Unscoped residue that must not look like a sibling.", "", "[]",
+        /*artifact=*/0, /*conversation_id=*/0);
+    s.create_entry(tid, "project", "other thread",
+        "Pinned elsewhere.", "", "[]", /*artifact=*/0, other_conv.id);
+
+    TenantStore::EntryFilter f;
+    f.conversation_id = conv.id;
+    f.exact_conversation = true;
+    auto rows = s.list_entries(tid, f);
+    REQUIRE(rows.size() == 1);
+    CHECK(rows[0].id == sibling.id);
+
+    // Recency LIMIT must not hide the sibling behind a flood of unscoped
+    // rows — the reason this is a SQL predicate, not a post-filter.
+    for (int i = 0; i < 20; ++i) {
+        s.create_entry(tid, "project", "flood " + std::to_string(i),
+            "Unscoped flood.", "", "[]");
+    }
+    f.limit = 5;
+    auto capped = s.list_entries(tid, f);
+    REQUIRE(capped.size() == 1);
+    CHECK(capped[0].id == sibling.id);
+
+    // FTS path uses the same exact predicate.
+    f.q = "sibling";
+    f.limit = 15;
+    auto hits = s.list_entries(tid, f);
+    REQUIRE(hits.size() == 1);
+    CHECK(hits[0].id == sibling.id);
+}
+
 TEST_CASE("graduated search prefers conversation hits, fills from tenant-wide") {
     TempDb db;
     TenantStore s;