Lineage Phase B1 — complete instrumentation (OSS/SQLite): update/hard_delete/status_change + reflection supersede + op=aggregate

.claude/rules/reflexio-patterns.md

@@ -23,3 +23,8 @@ paths:
 
 ## Config
 - `tool_can_use` lives at root `Config` level — shared across success evaluation and feedback extraction (NOT per-`AgentSuccessConfig`)
+
+## SQLite storage: lineage events & atomicity (one connection = one transaction)
+- The `sqlite3` connection is shared and `autocommit=False`, so `conn.commit()` **anywhere flushes the entire pending transaction**, not just the adjacent statement. The FTS/vec helpers (`_fts_*`, `_vec_*` in `sqlite_storage/_base.py`) **self-commit** internally.
+- **NEVER** interleave a self-committing helper between two writes you need atomic. E.g. `emit lineage event → _fts_delete()/_vec_delete() → DELETE row → commit` looks atomic ("one `with self._lock:` block, one commit at the end") but the helper's commit durably writes the audit event **before** the row is deleted — so a crash or a no-op delete leaves a phantom `hard_delete`/`status_change` event for a row that still exists. This bit a whole family of B1 delete/update methods.
+- **ALWAYS** emit a lineage event only **after** (or in the same `conn.commit()` as) the mutation it attests to, guarded on `cur.rowcount > 0`; run `_fts_*`/`_vec_*` cleanup **after** that commit (index maintenance, not the audited fact). The existence/eligibility check must use the **same predicate** (including `user_id` scope) as the mutation — otherwise you audit erasures that never happened, including for another user's row in the same org.

reflexio/models/api_schema/domain/entities.py

@@ -449,6 +449,9 @@ class LineageEvent(BaseModel):
     request_id: str = ""
     reason: str = ""
     created_at: int = 0
+    from_status: str | None = None
+    to_status: str | None = None
+    status_namespace: str | None = None
 
 
 class LineageContext(BaseModel):

reflexio/server/services/generation_service.py

@@ -299,7 +299,10 @@ def run(
             # profile/playbook context. Wrapped in a broad except so a
             # reflection bug never breaks the publish.
             self._maybe_run_reflection(
-                user_id=user_id, agent_version=agent_version, source=source
+                user_id=user_id,
+                request_id=request_id,
+                agent_version=agent_version,
+                source=source,
             )
 
             # Create generation services and requests
@@ -543,7 +546,7 @@ def _should_sample_group_evaluation(self, *, user_id: str, session_id: str) -> b
         )
 
     def _maybe_run_reflection(
-        self, *, user_id: str, agent_version: str, source: str | None
+        self, *, user_id: str, request_id: str, agent_version: str, source: str | None
     ) -> None:
         """Best-effort reflection pass before extraction.
 
@@ -558,6 +561,7 @@ def _maybe_run_reflection(
             service.run(
                 ReflectionServiceRequest(
                     user_id=user_id,
+                    request_id=request_id,
                     agent_version=agent_version,
                     source=source,
                 )

reflexio/server/services/playbook/playbook_aggregator.py

@@ -4,12 +4,14 @@
 import logging
 import os
 import time
+import uuid
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     import numpy as np
 
+from reflexio.models.api_schema.domain.entities import LineageEvent
 from reflexio.models.api_schema.service_schemas import (
     AgentPlaybook,
     AgentPlaybookSnapshot,
@@ -36,6 +38,7 @@
     ensure_playbook_content,
 )
 from reflexio.server.services.service_utils import log_model_response
+from reflexio.server.tracing import capture_anomaly
 from reflexio.server.usage_metrics import record_usage_event
 
 logger = logging.getLogger(__name__)
@@ -103,7 +106,7 @@ def _get_new_user_playbooks_count(
         # Count user playbooks with ID greater than last processed using efficient count query
         # Only count current user playbooks (status=None), not archived or pending ones.
         # Singleton aggregation operates on the user's whole playbook set — no name filter.
-        new_count = self.storage.count_user_playbooks(  # type: ignore[reportOptionalMemberAccess]
+        new_count = self.storage.count_user_playbooks(  # pyright: ignore[reportOptionalMemberAccess]
             min_user_playbook_id=last_processed_id,
             agent_version=self.agent_version,
             status_filter=[None],
@@ -572,6 +575,8 @@ def run(self, playbook_aggregator_request: PlaybookAggregatorRequest) -> dict:
             dict: Aggregation stats with keys: clusters_found, user_playbooks_processed, playbooks_generated, skipped (optional)
         """
         aggregation_start = time.perf_counter()
+        # Stable id for this aggregation run — groups all lineage events produced below.
+        _run_id = str(uuid.uuid4())
         _empty_stats = {
             "clusters_found": 0,
             "user_playbooks_processed": 0,
@@ -875,6 +880,40 @@ def run(self, playbook_aggregator_request: PlaybookAggregatorRequest) -> dict:
                             )
                         ],
                     )
+                    # Emit set-level aggregate lineage event (W3C PROV wasDerivedFrom, M:N).
+                    # Best-effort (PB-7): save_agent_playbooks already committed; a transient
+                    # append failure must NOT abort the run. Gap is acceptable for B1 since
+                    # the legacy change log remains the source of record until B3.
+                    member_ids = [
+                        str(fb.user_playbook_id)
+                        for fb in cluster_playbooks
+                        if fb.user_playbook_id
+                    ]
+                    try:
+                        self.storage.append_lineage_event(  # pyright: ignore[reportOptionalMemberAccess]
+                            LineageEvent(
+                                org_id=self.request_context.org_id,
+                                entity_type="agent_playbook",
+                                entity_id=str(saved_fb.agent_playbook_id),
+                                op="aggregate",
+                                prov_relation="wasDerivedFrom",
+                                source_ids=member_ids,
+                                actor="aggregator",
+                                request_id=_run_id,
+                                reason="user->agent aggregation",
+                            )
+                        )
+                    except Exception:  # noqa: BLE001
+                        logger.warning(
+                            "aggregate lineage event failed for agent_playbook %s",
+                            saved_fb.agent_playbook_id,
+                            exc_info=True,
+                        )
+                        capture_anomaly(
+                            "lineage.aggregate.append_failed",
+                            entity_id=str(saved_fb.agent_playbook_id),
+                            org_id=self.request_context.org_id,
+                        )
 
             # Store fingerprints in operation state
             mgr.update_cluster_fingerprints(

reflexio/server/services/reflection/reflection_service.py

@@ -34,6 +34,7 @@
 from reflexio.models.api_schema.domain.entities import (
     Citation,
     Interaction,
+    LineageContext,
     UserPlaybook,
     UserProfile,
 )
@@ -447,15 +448,19 @@ def _replace_profile(
         decision: ReflectionDecision,
         cited: UserProfile,
     ) -> bool:
-        """Insert the replacement profile, then archive the cited row.
+        """Insert the replacement profile, then supersede the cited row.
 
         Insert-first ordering means that if ``add_user_profile`` raises,
         the cited row stays current and the per-decision exception
         handler reports ``failed_count``. Only after the new row is
-        durable do we flip the cited row to ARCHIVED — and if *that*
-        fails we log at ERROR rather than silently dropping the user's
-        data, leaving a transient duplicate that downstream dedup can
-        clean up.
+        durable do we atomically supersede the cited row via
+        ``supersede_record``, which sets ``status=SUPERSEDED`` with a
+        ``superseded_by`` pointer and appends a ``revise`` lineage event.
+        If the CAS guard fails (incumbent no longer CURRENT), the
+        just-inserted successor is deleted (not audited); if that delete
+        itself raises the exception propagates to the caller. If
+        ``supersede_record`` raises, log at ERROR and accept a transient
+        duplicate rather than silently dropping user data.
         """
         storage = self.request_context.storage
         if storage is None:
@@ -482,14 +487,22 @@ def _replace_profile(
             new_content=new_profile.content,
         )
         storage.add_user_profile(cited.user_id, [new_profile])
+        ctx = LineageContext(
+            op_kind="revise",
+            actor="reflection",
+            request_id=request.request_id,
+        )
         try:
-            archived = storage.archive_profile_by_id(
-                user_id=request.user_id, profile_id=cited.profile_id
+            superseded = storage.supersede_record(
+                entity_type="profile",
+                incumbent_id=str(cited.profile_id),
+                successor_id=str(new_profile.profile_id),
+                context=ctx,
             )
         except Exception as exc:  # noqa: BLE001
             with sentry_tags(
                 subsystem="reflection",
-                op="archive_after_insert",
+                op="supersede_after_insert",
                 kind="profile",
                 org_id=self.request_context.org_id,
                 user_id=cited.user_id,
@@ -504,7 +517,12 @@ def _replace_profile(
                     new_profile.profile_id,
                 )
             return True
-        if not archived:
+        if not superseded:
+            # lost the CAS race: incumbent was no longer CURRENT — drop the
+            # just-added successor (never live) without emitting an audit event.
+            storage.delete_profiles_by_ids(
+                [new_profile.profile_id], emit_hard_delete=False
+            )
             with sentry_tags(
                 subsystem="reflection",
                 op="archive_after_insert_noop",

reflexio/server/services/reflection/reflection_service_utils.py

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import uuid
 from typing import Literal
 
 from pydantic import BaseModel, Field
@@ -29,6 +30,11 @@ class ReflectionServiceRequest(BaseModel):
 
     Args:
         user_id (str): User to scope the bookmark and window to.
+        request_id (str): The publish pass's own request id; used as the
+            lineage event ``request_id`` on revise events so B3
+            reconstruction can link revisions back to the triggering pass.
+            Defaults to a fresh UUID hex so two passes on the same profile
+            with no explicit request_id produce distinct lineage events.
         agent_version (str): Agent version of the current publish; copied
             into replacement playbooks.
         source (str | None): Optional source filter for the window.
@@ -37,6 +43,7 @@ class ReflectionServiceRequest(BaseModel):
     """
 
     user_id: str
+    request_id: str = Field(default_factory=lambda: uuid.uuid4().hex)
     agent_version: str = ""
     source: str | None = None
 

reflexio/server/services/storage/sqlite_storage/_base.py

@@ -1247,22 +1247,34 @@ def _migrate_lineage_event_table(self) -> None:
         with self._lock:
             self.conn.executescript("""
                 CREATE TABLE IF NOT EXISTS lineage_event (
-                    event_id      INTEGER PRIMARY KEY AUTOINCREMENT,
-                    org_id        TEXT NOT NULL,
-                    entity_type   TEXT NOT NULL,
-                    entity_id     TEXT NOT NULL,
-                    op            TEXT NOT NULL,
-                    prov_relation TEXT NOT NULL DEFAULT '',
-                    source_ids    TEXT NOT NULL DEFAULT '[]',
-                    actor         TEXT NOT NULL DEFAULT '',
-                    request_id    TEXT NOT NULL DEFAULT '',
-                    reason        TEXT NOT NULL DEFAULT '',
-                    created_at    INTEGER NOT NULL,
+                    event_id         INTEGER PRIMARY KEY AUTOINCREMENT,
+                    org_id           TEXT NOT NULL,
+                    entity_type      TEXT NOT NULL,
+                    entity_id        TEXT NOT NULL,
+                    op               TEXT NOT NULL,
+                    prov_relation    TEXT NOT NULL DEFAULT '',
+                    source_ids       TEXT NOT NULL DEFAULT '[]',
+                    actor            TEXT NOT NULL DEFAULT '',
+                    request_id       TEXT NOT NULL DEFAULT '',
+                    reason           TEXT NOT NULL DEFAULT '',
+                    created_at       INTEGER NOT NULL,
                     UNIQUE (org_id, entity_type, entity_id, op, request_id)
                 );
                 CREATE INDEX IF NOT EXISTS idx_lineage_entity
                     ON lineage_event (entity_type, entity_id);
             """)
+            existing_cols = {
+                row["name"]
+                for row in self.conn.execute(
+                    "PRAGMA table_info(lineage_event)"
+                ).fetchall()
+            }
+            for col in ("from_status", "to_status", "status_namespace"):
+                if col not in existing_cols:
+                    self.conn.execute(
+                        f"ALTER TABLE lineage_event ADD COLUMN {col} TEXT"  # noqa: S608
+                    )
+                    logger.info("Added %s column to lineage_event", col)
             self.conn.commit()
 
     def _migrate_agent_playbook_source_windows(self) -> None:
@@ -2126,17 +2138,20 @@ def clear_user_data(self, user_id: str) -> dict[str, int]:
 -- ============================================================================
 
 CREATE TABLE IF NOT EXISTS lineage_event (
-    event_id      INTEGER PRIMARY KEY AUTOINCREMENT,
-    org_id        TEXT NOT NULL,
-    entity_type   TEXT NOT NULL,
-    entity_id     TEXT NOT NULL,
-    op            TEXT NOT NULL,
-    prov_relation TEXT NOT NULL DEFAULT '',
-    source_ids    TEXT NOT NULL DEFAULT '[]',
-    actor         TEXT NOT NULL DEFAULT '',
-    request_id    TEXT NOT NULL DEFAULT '',
-    reason        TEXT NOT NULL DEFAULT '',
-    created_at    INTEGER NOT NULL,
+    event_id         INTEGER PRIMARY KEY AUTOINCREMENT,
+    org_id           TEXT NOT NULL,
+    entity_type      TEXT NOT NULL,
+    entity_id        TEXT NOT NULL,
+    op               TEXT NOT NULL,
+    prov_relation    TEXT NOT NULL DEFAULT '',
+    source_ids       TEXT NOT NULL DEFAULT '[]',
+    actor            TEXT NOT NULL DEFAULT '',
+    request_id       TEXT NOT NULL DEFAULT '',
+    reason           TEXT NOT NULL DEFAULT '',
+    created_at       INTEGER NOT NULL,
+    from_status      TEXT,
+    to_status        TEXT,
+    status_namespace TEXT,
     UNIQUE (org_id, entity_type, entity_id, op, request_id)
 );
 CREATE INDEX IF NOT EXISTS idx_lineage_entity ON lineage_event (entity_type, entity_id);

reflexio/server/services/storage/sqlite_storage/_lineage.py

@@ -41,6 +41,9 @@ def _append_event_stmt(
     request_id: str,
     reason: str,
     created_at: int | None = None,
+    from_status: str | None = None,
+    to_status: str | None = None,
+    status_namespace: str | None = None,
 ) -> sqlite3.Cursor:
     """Insert a lineage event row; no-ops on (org_id, entity_type, entity_id, op, request_id) duplicate.
 
@@ -49,8 +52,9 @@ def _append_event_stmt(
     return conn.execute(
         "INSERT OR IGNORE INTO lineage_event "
         "(org_id, entity_type, entity_id, op, prov_relation, source_ids, "
-        "actor, request_id, reason, created_at) "
-        "VALUES (?,?,?,?,?,?,?,?,?,?)",
+        "actor, request_id, reason, created_at, "
+        "from_status, to_status, status_namespace) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
         (
             org_id,
             entity_type,
@@ -62,6 +66,9 @@ def _append_event_stmt(
             request_id,
             reason,
             created_at if created_at is not None else int(time.time()),
+            from_status,
+            to_status,
+            status_namespace,
         ),
     )
 
@@ -100,6 +107,9 @@ def append_lineage_event(self, event: LineageEvent) -> int:
                 request_id=event.request_id,
                 reason=event.reason,
                 created_at=created,
+                from_status=event.from_status,
+                to_status=event.to_status,
+                status_namespace=event.status_namespace,
             )
             if (
                 cur.rowcount == 0
@@ -168,6 +178,9 @@ def get_lineage_events(
                 request_id=r["request_id"],
                 reason=r["reason"],
                 created_at=r["created_at"],
+                from_status=r["from_status"],
+                to_status=r["to_status"],
+                status_namespace=r["status_namespace"],
             )
             for r in rows
         ]