fix: batch dedupe must ignore TTL expiry (zero-loss); scope depends_on edges on task create (security)

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+Zero-loss fix: batch dedupe no longer breaks once an item's TTL expires. `existing_source_ids()` (the set that makes `POST /ingest/batch` idempotent) was built on `_load_active_rows`, which the earlier `forget_after` work taught to hide expired rows. So once a batch item's `forget_after` passed, its `source_id` dropped out of the dedup set and re-POSTing the same id re-ingested it, writing a second archive and vector row on every re-POST (unbounded duplication of hidden-but-present content). Dedup now reads all physically-present rows via a new `include_expired` path on `_load_active_rows`: a TTL-expired row is only hidden from recall, it still exists on disk, so its id still dedupes. Superseded rows (`valid_to` set) stay excluded from the set, so re-importing intentionally-cleared content still re-adds it, and recall-time TTL behavior is unchanged (search still hides expired rows).
+
+Security fix: `POST /tasks` with `depends_on` now enforces the same project scope as the edge endpoints. Create was the one graph-mutating path that skipped `_enforce_edge_project_scope`: it applied token binding but then passed `depends_on` straight into `create_task`, which makes a blocks edge with no scope check. A project-scoped registry token could therefore create cross-project blocks edges and enumerate foreign task existence (a real id returned 200, a bogus id returned a 400 naming the missing task). When a token binds a project, each `depends_on` id is now checked to belong to that project before the task is created, returning the same non-enumerating 403 the edge endpoints use (foreign and nonexistent ids are indistinguishable). Tokenless and standalone behavior is unchanged.
+
 Zero-loss fix: `TemporalKnowledgeGraph.add_entity` no longer silently overwrites an existing entity's type or properties. Because `add_triple` re-adds every subject/object on each write, the same entity is constantly re-inserted with whatever type the current extraction guessed (frequently the `unknown` placeholder, or a conflicting per-mention guess). The old `ON CONFLICT(id) DO UPDATE` was last-writer-wins, so a concrete classification (for example the `agent`/`lesson` types written by `crystallize`) was clobbered back to `unknown` by a later default-typed mention, and a re-added name flipped the display casing, both with no record. Triple relationships were already tombstoned via `valid_from`/`valid_to`, but entity attributes were mutated in place. `add_entity` is now non-destructive: it keeps the first-seen `name`, keeps the first-seen concrete `type` and only upgrades the `unknown` placeholder to a concrete type (enrichment, never a downgrade), merges `properties` additively with existing values winning on a key clash, and preserves the original `created_at`. Properties were always the `{}` default in practice (no caller passes them today), so that arm is forward-safety; the type and name overwrites were live. Deliberate re-classification with old-value history would need the triple-style temporal treatment, which no caller requires, so the lightweight merge is used rather than a heavier versioning table.
 
 TTL fix: `forget_after` supplied via `ingest_batch` now actually expires. The retrieval-time TTL filter read `forget_after` only from the top level of a vector row's metadata, but `ingest_batch` nests the caller's per-item metadata under `meta["metadata"]` (alongside the top-level `agent` tag), so a `forget_after` passed through `POST /ingest/batch` or `ingest_batch(items=[{"metadata": {"forget_after": ...}}])` never hid anything and the row stayed visible forever. The filter now reads `forget_after` from both the top level and the nested user-metadata dict (top level wins if both are set), so expiry works identically whether the row came from a flat single `ingest` / low-level `add` or from batch ingest (including reconcile-repaired batch rows, which nest the same way). Zero-loss is unchanged: expired rows are only hidden from recall, never deleted. The prior batch TTL test hand-built a flat row shape that the real batch path never produces, which masked the bug; it now uses the true nested shape and a new end-to-end test drives `ingest_batch` directly.

taosmd/http_server.py

@@ -696,8 +696,16 @@ def _apply_token_binding(self, identity: str | None, project: str | None) -> tup
             deferred identity-keying work, not this layer.
 
             Returns ``(resolved_project, ok)`` where ``ok=False`` means the
-            response has already been written and the caller must return.
+            response has already been written and the caller must return. The
+            token-derived project (the verified ``project_id`` claim, or
+            ``None`` for a tokenless or global-token request) is also stashed
+            on ``self._token_project`` for handlers that must scope purely on
+            the token, never on the body-supplied project (see
+            ``_handle_task_create``'s ``depends_on`` guard). It is reset to
+            ``None`` on every call so a keep-alive connection cannot leak a
+            prior request's binding.
             """
+            self._token_project = None
             if _registry_verifier is None:
                 return project, True
             auth = self.headers.get("Authorization", "")
@@ -726,6 +734,7 @@ def _apply_token_binding(self, identity: str | None, project: str | None) -> tup
                 self._send_json(403, {"error": f"registry auth: {exc}"})
                 return None, False
             verified_project = claims.get("project_id")
+            self._token_project = verified_project
             if verified_project is not None:
                 project = verified_project
             # Token proves identity; a grant proves permission. Any verified
@@ -1337,12 +1346,27 @@ def _handle_task_create(self) -> None:
             project, ok = self._apply_token_binding(created_by, project)
             if not ok:
                 return
+            token_project = self._token_project
             try:
                 priority = int(priority)
             except (TypeError, ValueError) as exc:
                 raise _BadRequest("'priority' must be an integer") from exc
             if depends_on is not None and not isinstance(depends_on, list):
                 raise _BadRequest("'depends_on' must be a list of task IDs when provided")
+            # ``depends_on`` becomes a blocks edge (tasks.create_task), so it
+            # must obey the same non-enumerating project scope as the /edges
+            # endpoints: a token bound to a project may only depend on that
+            # project's tasks. Foreign and nonexistent ids yield the identical
+            # 403 so task existence cannot be probed. The scope keys on the
+            # TOKEN-bound project (like the /edges handlers, which derive scope
+            # purely from the token), NOT the post-binding ``project`` -- for a
+            # tokenless caller that is just the body-supplied tag and must not
+            # restrict anything. Tokenless / standalone / global-token requests
+            # (``token_project is None``) pass through untouched.
+            if token_project is not None and depends_on:
+                for dep_id in depends_on:
+                    if not self._enforce_edge_project_scope(token_project, dep_id, dep_id):
+                        return
             result = runner.run(
                 service.task_create(
                     title,

taosmd/vector_memory.py

@@ -741,6 +741,7 @@ def _load_active_rows(
         project: str | None = None,
         search_agents: list[str] | None = None,
         now: float | None = None,
+        include_expired: bool = False,
     ):
         """Fetch active (non-superseded) rows, scoped by project/agent.
 
@@ -754,6 +755,12 @@ def _load_active_rows(
         Non-numeric or missing ``forget_after`` values are silently ignored so
         existing memories are never affected. ``now`` defaults to
         ``time.time()``; pass an explicit float in tests to control the clock.
+
+        ``include_expired=True`` skips the TTL filter only: superseded rows
+        (``valid_to`` set) stay excluded, but rows hidden purely because their
+        ``forget_after`` has passed are returned. Batch dedupe uses this so an
+        expired-but-still-present id is not re-ingested (see
+        ``existing_source_ids``). Recall paths never pass it.
         """
         if now is None:
             now = time.time()
@@ -782,21 +789,22 @@ def _load_active_rows(
             # nest the caller's per-item metadata under ``meta["metadata"]``.
             # Tolerating both shapes makes forget_after expire identically no
             # matter which ingest path wrote the row. Top level wins if set.
-            fa = meta.get("forget_after")
-            if fa is None:
-                inner = meta.get("metadata")
-                if isinstance(inner, dict):
-                    fa = inner.get("forget_after")
-            if fa is not None:
-                try:
-                    if float(fa) < now:
-                        continue
-                except (TypeError, ValueError):
-                    logger.debug(
-                        "ignore non-numeric forget_after=%r on row id=%s",
-                        fa,
-                        row["id"],
-                    )
+            if not include_expired:
+                fa = meta.get("forget_after")
+                if fa is None:
+                    inner = meta.get("metadata")
+                    if isinstance(inner, dict):
+                        fa = inner.get("forget_after")
+                if fa is not None:
+                    try:
+                        if float(fa) < now:
+                            continue
+                    except (TypeError, ValueError):
+                        logger.debug(
+                            "ignore non-numeric forget_after=%r on row id=%s",
+                            fa,
+                            row["id"],
+                        )
 
             if project is not None or search_agents is not None:
                 # Project filter: skip rows positively tagged with a different
@@ -821,9 +829,18 @@ def existing_source_ids(self, agent: str | None = None) -> set[str]:
         whose ``id`` is already present are skipped instead of duplicated.
         Rows tagged with a different agent are excluded when ``agent`` is set;
         untagged rows are included, matching the search-scoping rules.
+
+        Dedup must consider every physically-present row, including rows whose
+        ``forget_after`` has passed (``include_expired=True``): a TTL-expired
+        row is only hidden from recall, it still exists on disk, so re-ingesting
+        its id would write a second archive/vector row every re-POST (zero-loss
+        violation). Superseded rows (``valid_to`` set) are genuinely removed and
+        stay excluded, so re-importing intentionally-cleared content re-adds it.
         """
         out: set[str] = set()
-        for row in self._load_active_rows(search_agents=[agent] if agent else None):
+        for row in self._load_active_rows(
+            search_agents=[agent] if agent else None, include_expired=True
+        ):
             try:
                 meta = json.loads(row["metadata_json"])
             except (json.JSONDecodeError, TypeError):

tests/test_http_server_registry_auth.py

@@ -540,6 +540,75 @@ def test_edge_endpoints_tokenless_on_authed_server_unchanged(project_server):
     assert body["removed_ts"] is not None
 
 
+def test_task_create_scoped_token_cross_project_depends_on_is_403(project_server):
+    """A proj-a token cannot create a task that depends_on a proj-b task.
+
+    ``depends_on`` becomes a blocks edge, so it must obey the same
+    project scope as the /edges endpoints."""
+    t_foreign = _create_task(project_server, "Foreign blocker", project="proj-b")
+    tok = _make_token("agent-1", project_id="proj-a", iss=registry_auth.REGISTRY_ISS)
+    status, _ = _post_json(
+        project_server, "/tasks",
+        {"title": "my proj-a task", "created_by": "agent-1",
+         "depends_on": [t_foreign]},
+        token=tok)
+    assert status == 403
+    # The foreign task must not have been blocked by the phantom edge.
+    assert t_foreign in _ready_ids(project_server)
+
+
+def test_task_create_scoped_token_depends_on_does_not_enumerate(project_server):
+    """A foreign existing depends_on target and a nonexistent one must be
+    indistinguishable (same 403 status, same error body)."""
+    t_foreign = _create_task(project_server, "Foreign target", project="proj-b")
+    tok = _make_token("agent-1", project_id="proj-a", iss=registry_auth.REGISTRY_ISS)
+    s_foreign, b_foreign = _post_json(
+        project_server, "/tasks",
+        {"title": "probe foreign", "created_by": "agent-1",
+         "depends_on": [t_foreign]},
+        token=tok)
+    s_missing, b_missing = _post_json(
+        project_server, "/tasks",
+        {"title": "probe missing", "created_by": "agent-1",
+         "depends_on": ["t-000000000000"]},
+        token=tok)
+    assert s_foreign == s_missing == 403
+    assert b_foreign == b_missing
+
+
+def test_task_create_scoped_token_same_project_depends_on_succeeds(project_server):
+    """A proj-a token can create a task depending on a proj-a task."""
+    t_blocker = _create_task(project_server, "Local blocker", project="proj-a")
+    tok = _make_token("agent-1", project_id="proj-a", iss=registry_auth.REGISTRY_ISS)
+    status, body = _post_json(
+        project_server, "/tasks",
+        {"title": "local blocked", "created_by": "agent-1",
+         "depends_on": [t_blocker]},
+        token=tok)
+    assert status == 200, body
+    new_id = body["id"]
+    # The blocks edge is real: the new task is not ready while the blocker is open.
+    assert new_id not in _ready_ids(project_server)
+
+
+def test_task_create_tokenless_depends_on_unchanged(project_server):
+    """Without a token, create-with-depends_on works exactly as standalone,
+    even across projects.
+
+    The scope guard keys on the token-bound project, not the body-supplied
+    ``project`` tag: a tokenless caller may tag its own task one project and
+    depend on a task in another, exactly as on the unauthed/standalone path.
+    Using a genuinely CROSS-project dep (blocker in proj-a, new task tagged
+    proj-b) proves tokenless is truly unrestricted, not merely same-project."""
+    t_blocker = _create_task(project_server, "Plain blocker", project="proj-a")
+    status, body = _post_json(
+        project_server, "/tasks",
+        {"title": "plain blocked", "created_by": "setup",
+         "project": "proj-b", "depends_on": [t_blocker]})
+    assert status == 200, body
+    assert body["id"] not in _ready_ids(project_server)
+
+
 def test_task_update_scoped_token_cannot_touch_foreign_project(project_server):
     """A proj-a token must not mutate a proj-b task, and the refusal must
     not reveal whether the foreign task exists."""

tests/test_ttl_filter.py

@@ -286,6 +286,61 @@ def test_ingest_batch_forget_after_expires_via_api(isolated):
     assert total == 3
 
 
+def test_ingest_batch_expired_item_still_dedupes(isolated):
+    """A batch item whose forget_after has passed must still dedupe on re-POST.
+
+    An expired row is only hidden from recall; it is physically present, so a
+    re-import of the same id must be skipped (idempotency) rather than writing
+    a second archive/vector row. Regression guard for the #188 TTL filter
+    leaking into existing_source_ids() and breaking batch dedupe (zero-loss:
+    unbounded archive duplication on every re-POST of an expired id)."""
+    stores = asyncio.run(taosmd_api._ensure_stores(str(isolated)))
+    _patch_embedder(stores)
+    agent = "ttl-dedupe-agent"
+    vmem = stores["vector"]
+
+    past = time.time() - 3600
+    item = [{"text": "stale note", "id": "stale-1",
+             "metadata": {"forget_after": past}}]
+
+    r1 = asyncio.run(taosmd.ingest_batch(item, agent=agent, data_dir=str(isolated)))
+    assert r1["ingested"] == 1 and r1["skipped"] == 0, r1
+
+    r2 = asyncio.run(taosmd.ingest_batch(item, agent=agent, data_dir=str(isolated)))
+    assert r2["ingested"] == 0, "expired item must not re-ingest"
+    assert r2["skipped"] == 1, "expired item id must be deduped on re-POST"
+
+    # Zero-loss / no duplication: exactly one physical row for the id.
+    total = vmem._conn.execute("SELECT COUNT(*) FROM vector_memory").fetchone()[0]
+    assert total == 1, "re-POST of an expired id must not write a second row"
+
+
+def test_existing_source_ids_includes_expired_rows(tmp_path):
+    """existing_source_ids() must report source_ids of expired rows too.
+
+    The set backs batch dedupe; an expired row still exists on disk, so its
+    source_id must dedupe. A non-expired row is reported as before."""
+    vmem = _make_store(tmp_path)
+    try:
+        past = time.time() - 3600
+        future = time.time() + 86400
+        asyncio.run(vmem.add(
+            "expired sourced row",
+            metadata={"agent": "a", "metadata": {"source_id": "expired-sid",
+                                                 "forget_after": past}}))
+        asyncio.run(vmem.add(
+            "active sourced row",
+            metadata={"agent": "a", "metadata": {"source_id": "active-sid",
+                                                 "forget_after": future}}))
+
+        sids = vmem.existing_source_ids(agent="a")
+        assert "active-sid" in sids
+        assert "expired-sid" in sids, (
+            "expired-but-present row must still dedupe in existing_source_ids()")
+    finally:
+        asyncio.run(vmem.close())
+
+
 def test_ttl_filter_uses_now_override(tmp_path):
     """Passing an explicit now value controls which rows are expired."""
     vmem = _make_store(tmp_path)