Lineage Phase B2 — TTL GC (tombstone garbage collection), OSS/SQLite - #188
Merged
Conversation
…istent TEXT age compare (B2 Task 2 review)
…gated, sentry checkpoints) — B2 Task 3
…e fallback + mid-tick stop (B2 Task 3 review)
📝 WalkthroughWalkthroughAdds an opt-in lineage tombstone garbage-collection system. A new ChangesLineage Tombstone GC Feature
Sequence Diagram(s)sequenceDiagram
participant Lifespan as app lifespan
participant GCS as LineageGCScheduler
participant BootCtx as bootstrap RequestContext
participant OrgCtx as per-org RequestContext
participant SQLite as SQLiteLineageMixin
Lifespan->>GCS: maybe_start_lineage_gc(factory, bootstrap_org_id)
GCS->>BootCtx: read lineage_gc config
GCS->>GCS: start() — daemon thread
loop _run_loop every poll_interval_seconds
GCS->>BootCtx: storage.list_org_ids()
BootCtx-->>GCS: [org_id, ...]
loop per org_id
GCS->>OrgCtx: request_context_factory(org_id)
OrgCtx-->>GCS: config + storage
loop per entity_type
GCS->>SQLite: gc_expired_tombstones(entity_type, older_than_epoch)
SQLite->>SQLite: SELECT candidates (status + age cutoff)
SQLite->>SQLite: _is_on_legal_hold check
SQLite->>SQLite: emit hard_delete lineage event
SQLite->>SQLite: DELETE FTS/vector + base row
SQLite-->>GCS: deleted_count
end
GCS->>GCS: anomaly if total > _HIGH_VOLUME_THRESHOLD
end
end
Lifespan->>GCS: stop()
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (3)
tests/models/test_lineage_gc_config.py (1)
10-41: ⚡ Quick winAdd invalid-value tests for GC timing fields.
This suite covers happy paths but not bad inputs. Please add
ValidationErrortests for negativetombstone_grace_window_daysand non-positivepoll_interval_secondsto lock in safety guarantees.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/models/test_lineage_gc_config.py` around lines 10 - 41, Add validation tests for invalid GC timing field values in the LineageGCConfig class. Create two new test functions: one that verifies a ValidationError is raised when tombstone_grace_window_days is set to a negative value, and another that verifies a ValidationError is raised when poll_interval_seconds is set to zero or a negative value. These tests should instantiate LineageGCConfig with invalid parameter values and assert that ValidationError is raised to ensure the model validates these fields properly.tests/server/services/storage/test_storage_contract_gc_tombstones.py (1)
72-179: ⚡ Quick winAdd backend-agnostic contract coverage for
ARCHIVEDeligibility.
ARCHIVEDdeletion is currently asserted only in SQLite integration tests. Add a contract test (profile status set toARCHIVED) so all backends must honor the same GC eligibility set.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_storage_contract_gc_tombstones.py` around lines 72 - 179, Add a new contract test function that validates ARCHIVED profile deletion by GC. Create a test (similar in structure to test_gc_deletes_aged_merged_profile_and_emits_hard_delete) that creates a profile with ARCHIVED status using an old timestamp, calls gc_expired_tombstones with an appropriate cutoff, and verifies the ARCHIVED profile is deleted and a hard_delete lineage event is emitted. This ensures all storage backends enforce the same GC eligibility rules for ARCHIVED status, not just SQLite.reflexio/server/api.py (1)
2978-2985: ⚡ Quick winResolve the lifespan bootstrap org once and reuse it for both schedulers.
Calling
_lifespan_org_id()twice can yield inconsistent bootstrap scopes if a custom resolver is non-deterministic or flaky.♻️ Suggested refactor
if mount_data_plane: validate_llm_availability() from reflexio.server.llm.rerank import prewarm as _prewarm_cross_encoder _prewarm_cross_encoder() + lifespan_org_id = _lifespan_org_id() # The scheduler discovers every org with resumable work each tick and # drives a per-org worker with org-scoped claims, so it is not limited # to the bootstrap org. The bootstrap org is only used to read config # and to seed cross-org discovery. scheduler = maybe_start_resume_scheduler( lambda org_id: RequestContext(org_id=org_id), - bootstrap_org_id=_lifespan_org_id(), + bootstrap_org_id=lifespan_org_id, ) gc_scheduler = maybe_start_lineage_gc( lambda org_id: RequestContext(org_id=org_id), - bootstrap_org_id=_lifespan_org_id(), + bootstrap_org_id=lifespan_org_id, )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/api.py` around lines 2978 - 2985, The function _lifespan_org_id() is being called twice in the scheduler initialization block, which could result in inconsistent bootstrap org values being passed to maybe_start_resume_scheduler and maybe_start_lineage_gc if the function is non-deterministic. Call _lifespan_org_id() once before both scheduler calls, store the result in a variable, and then pass that same variable as the bootstrap_org_id argument to both maybe_start_resume_scheduler and maybe_start_lineage_gc to ensure consistency.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.superpowers/sdd/task-3-report.md:
- Line 12: The documentation in the report for `_discover_org_ids` states that
it falls back on both `NotImplementedError` and `AttributeError`, but the actual
implementation only handles `NotImplementedError`. Update the report text to
accurately reflect the current implementation behavior by removing the reference
to `AttributeError` from the fallback documentation, ensuring the reported
behavior matches what the code actually implements.
In `@reflexio/models/config_schema.py`:
- Around line 684-685: The LineageGCConfig model has two timing fields,
tombstone_grace_window_days and poll_interval_seconds, that currently lack
bounds validation, allowing negative or zero values which can cause safety
issues or tight scheduler loops. Add field validators to these fields to ensure
they only accept positive integer values (greater than zero), constraining
validation at the model level to prevent invalid configurations from being
accepted.
In `@reflexio/server/services/lineage/gc_scheduler.py`:
- Around line 136-141: The poll_interval variable read from
cfg.lineage_gc.poll_interval_seconds is used directly in self._stop_event.wait()
without validation, which means a zero or negative value will cause a busy-wait
loop that degrades worker availability. After retrieving poll_interval from the
configuration, add a clamp operation to ensure it has a minimum safe positive
value (e.g., use max() to ensure it is at least some reasonable threshold like 1
second) before passing it to self._stop_event.wait().
In `@reflexio/server/services/storage/sqlite_storage/_lineage.py`:
- Around line 345-389: The gc_expired_tombstones method accepts a limit
parameter that is documented as a maximum rows cap, but it does not validate
that the limit value is positive before using it in the SQL query construction.
Add validation at the beginning of the method (after the entity_type validation)
to ensure that limit is greater than zero, raising a ValueError with an
appropriate message if limit is non-positive (zero or negative). This will
preserve bounded and predictable garbage collection behavior.
- Around line 391-470: Wrap the entire operation block (from the
_append_event_stmt call through the self.conn.execute DELETE statement and
self.conn.commit) in a try-except block to ensure atomicity. In the except
clause, call self.conn.rollback() to undo any partial writes if an exception
occurs at any step, then re-raise the exception. This prevents partial writes
from persisting when unrelated operations commit later.
---
Nitpick comments:
In `@reflexio/server/api.py`:
- Around line 2978-2985: The function _lifespan_org_id() is being called twice
in the scheduler initialization block, which could result in inconsistent
bootstrap org values being passed to maybe_start_resume_scheduler and
maybe_start_lineage_gc if the function is non-deterministic. Call
_lifespan_org_id() once before both scheduler calls, store the result in a
variable, and then pass that same variable as the bootstrap_org_id argument to
both maybe_start_resume_scheduler and maybe_start_lineage_gc to ensure
consistency.
In `@tests/models/test_lineage_gc_config.py`:
- Around line 10-41: Add validation tests for invalid GC timing field values in
the LineageGCConfig class. Create two new test functions: one that verifies a
ValidationError is raised when tombstone_grace_window_days is set to a negative
value, and another that verifies a ValidationError is raised when
poll_interval_seconds is set to zero or a negative value. These tests should
instantiate LineageGCConfig with invalid parameter values and assert that
ValidationError is raised to ensure the model validates these fields properly.
In `@tests/server/services/storage/test_storage_contract_gc_tombstones.py`:
- Around line 72-179: Add a new contract test function that validates ARCHIVED
profile deletion by GC. Create a test (similar in structure to
test_gc_deletes_aged_merged_profile_and_emits_hard_delete) that creates a
profile with ARCHIVED status using an old timestamp, calls gc_expired_tombstones
with an appropriate cutoff, and verifies the ARCHIVED profile is deleted and a
hard_delete lineage event is emitted. This ensures all storage backends enforce
the same GC eligibility rules for ARCHIVED status, not just SQLite.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 8a9ec210-6d83-4373-adbe-c9377cb67432
📒 Files selected for processing (10)
.superpowers/sdd/task-3-report.mdreflexio/models/config_schema.pyreflexio/server/api.pyreflexio/server/services/lineage/gc_scheduler.pyreflexio/server/services/storage/sqlite_storage/_lineage.pyreflexio/server/services/storage/storage_base/_lineage.pytests/models/test_lineage_gc_config.pytests/server/services/lineage/test_gc_scheduler.pytests/server/services/storage/test_lineage_b2_gc_integration.pytests/server/services/storage/test_storage_contract_gc_tombstones.py
| - MOD: `reflexio/server/api.py` (lifespan wiring) | ||
|
|
||
| ## Deviations from spec | ||
| - `_discover_org_ids` uses `storage.list_org_ids()` (a generic cross-org sweep), falling back to bootstrap-only on `NotImplementedError`/`AttributeError`. The resume scheduler uses `list_resumable_work_org_ids` which is work-filtered. For GC, an unfiltered list is correct because we want to sweep all orgs regardless of whether they have pending work. |
There was a problem hiding this comment.
SDD mismatch: report claims AttributeError fallback that code does not implement.
The report says _discover_org_ids falls back on both NotImplementedError and AttributeError, but current code only handles NotImplementedError. Please align the report (or implementation) so behavior is documented accurately.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.superpowers/sdd/task-3-report.md at line 12, The documentation in the
report for `_discover_org_ids` states that it falls back on both
`NotImplementedError` and `AttributeError`, but the actual implementation only
handles `NotImplementedError`. Update the report text to accurately reflect the
current implementation behavior by removing the reference to `AttributeError`
from the fallback documentation, ensuring the reported behavior matches what the
code actually implements.
Comment on lines
+684
to
+685
| tombstone_grace_window_days: int = 90 | ||
| poll_interval_seconds: int = 86400 |
There was a problem hiding this comment.
Add bounds validation to GC timing fields.
LineageGCConfig currently accepts negative/zero timing values. Once enabled, this can create unsafe cutoffs or a tight scheduler loop. Constrain these at model validation time.
Proposed fix
class LineageGCConfig(BaseModel):
@@
- tombstone_grace_window_days: int = 90
- poll_interval_seconds: int = 86400
+ tombstone_grace_window_days: int = Field(default=90, ge=0)
+ poll_interval_seconds: int = Field(default=86400, gt=0)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| tombstone_grace_window_days: int = 90 | |
| poll_interval_seconds: int = 86400 | |
| tombstone_grace_window_days: int = Field(default=90, ge=0) | |
| poll_interval_seconds: int = Field(default=86400, gt=0) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/models/config_schema.py` around lines 684 - 685, The LineageGCConfig
model has two timing fields, tombstone_grace_window_days and
poll_interval_seconds, that currently lack bounds validation, allowing negative
or zero values which can cause safety issues or tight scheduler loops. Add field
validators to these fields to ensure they only accept positive integer values
(greater than zero), constraining validation at the model level to prevent
invalid configurations from being accepted.
Comment on lines
+136
to
+141
| poll_interval = cfg.lineage_gc.poll_interval_seconds | ||
| org_ids = self._discover_org_ids(bootstrap_ctx) | ||
| self._gc_tick(org_ids) | ||
| except Exception: | ||
| logger.exception("event=lineage_gc_scheduler_tick_failed") | ||
| self._stop_event.wait(poll_interval) |
There was a problem hiding this comment.
Clamp non-positive poll intervals to prevent a hot loop.
poll_interval_seconds is used directly in self._stop_event.wait(...); a 0 or negative value can spin the loop continuously and degrade worker availability.
💡 Suggested fix
cfg = bootstrap_ctx.configurator.get_config()
- poll_interval = cfg.lineage_gc.poll_interval_seconds
+ poll_interval = cfg.lineage_gc.poll_interval_seconds
+ if poll_interval <= 0:
+ logger.warning(
+ "event=lineage_gc_invalid_poll_interval "
+ "poll_interval_seconds=%s fallback_seconds=%s",
+ poll_interval,
+ _DEFAULT_POLL_INTERVAL_SECONDS,
+ )
+ poll_interval = _DEFAULT_POLL_INTERVAL_SECONDS
org_ids = self._discover_org_ids(bootstrap_ctx)
self._gc_tick(org_ids)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/services/lineage/gc_scheduler.py` around lines 136 - 141, The
poll_interval variable read from cfg.lineage_gc.poll_interval_seconds is used
directly in self._stop_event.wait() without validation, which means a zero or
negative value will cause a busy-wait loop that degrades worker availability.
After retrieving poll_interval from the configuration, add a clamp operation to
ensure it has a minimum safe positive value (e.g., use max() to ensure it is at
least some reasonable threshold like 1 second) before passing it to
self._stop_event.wait().
Comment on lines
+345
to
+389
| def gc_expired_tombstones( | ||
| self, *, entity_type: str, older_than_epoch: int, limit: int = 1000 | ||
| ) -> int: | ||
| """Hard-delete tombstone rows that are older than the given epoch cutoff. | ||
|
|
||
| Emits one ``hard_delete`` lineage event per deleted row, atomically, before | ||
| the DELETE commits. Rows on legal hold are skipped without emitting an event. | ||
|
|
||
| Args: | ||
| entity_type (str): One of ``"user_playbook"``, ``"agent_playbook"``, | ||
| or ``"profile"``. | ||
| older_than_epoch (int): Unix timestamp cutoff (exclusive). Rows whose | ||
| age column is strictly less than this value are eligible. | ||
| limit (int): Maximum rows to delete per call. Defaults to 1000. | ||
|
|
||
| Returns: | ||
| int: The number of rows physically deleted. | ||
|
|
||
| Raises: | ||
| ValueError: If ``entity_type`` is not a recognised entity type. | ||
| """ | ||
| meta = _GC_ENTITY_META.get(entity_type) | ||
| if meta is None: | ||
| raise ValueError(f"unknown entity_type: {entity_type!r}") | ||
| table, pk, age_col, age_is_text = meta | ||
|
|
||
| eligible_ph = ",".join("?" * len(_GC_ELIGIBLE_STATUSES)) | ||
| eligible_vals = list(_GC_ELIGIBLE_STATUSES) | ||
|
|
||
| if age_is_text: | ||
| # Use the same helper the writer uses so the cutoff string is | ||
| # byte-for-byte format-consistent with stored values. This keeps | ||
| # the ``<`` comparison truly exclusive (strict) at the boundary. | ||
| cutoff_iso = _epoch_to_iso(older_than_epoch) | ||
| select_sql = ( | ||
| f"SELECT {pk} FROM {table} " # noqa: S608 | ||
| f"WHERE status IN ({eligible_ph}) AND {age_col} < ? LIMIT ?" | ||
| ) | ||
| select_params: list[Any] = [*eligible_vals, cutoff_iso, limit] | ||
| else: | ||
| select_sql = ( | ||
| f"SELECT {pk} FROM {table} " # noqa: S608 | ||
| f"WHERE status IN ({eligible_ph}) AND {age_col} < ? LIMIT ?" | ||
| ) | ||
| select_params = [*eligible_vals, older_than_epoch, limit] |
There was a problem hiding this comment.
Validate limit before query construction.
limit is documented as a max rows cap, but non-positive values are currently accepted. Guard it explicitly to preserve bounded, predictable GC behavior.
Proposed fix
def gc_expired_tombstones(
self, *, entity_type: str, older_than_epoch: int, limit: int = 1000
) -> int:
@@
meta = _GC_ENTITY_META.get(entity_type)
if meta is None:
raise ValueError(f"unknown entity_type: {entity_type!r}")
+ if limit <= 0:
+ raise ValueError("limit must be > 0")
table, pk, age_col, age_is_text = meta🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/services/storage/sqlite_storage/_lineage.py` around lines 345
- 389, The gc_expired_tombstones method accepts a limit parameter that is
documented as a maximum rows cap, but it does not validate that the limit value
is positive before using it in the SQL query construction. Add validation at the
beginning of the method (after the entity_type validation) to ensure that limit
is greater than zero, raising a ValueError with an appropriate message if limit
is non-positive (zero or negative). This will preserve bounded and predictable
garbage collection behavior.
Comment on lines
+391
to
+470
| with self._lock: | ||
| rows = self.conn.execute(select_sql, select_params).fetchall() | ||
| if not rows: | ||
| return 0 | ||
|
|
||
| candidate_ids: list[str] = [str(r[0]) for r in rows] | ||
| ids_to_delete: list[str] = [] | ||
| for eid in candidate_ids: | ||
| if self._is_on_legal_hold(self.org_id, entity_type, eid): | ||
| capture_anomaly( | ||
| "lineage.gc.legal_hold_skip", | ||
| level="info", | ||
| org_id=self.org_id, | ||
| entity_type=entity_type, | ||
| entity_id=eid, | ||
| ) | ||
| continue | ||
| ids_to_delete.append(eid) | ||
|
|
||
| if not ids_to_delete: | ||
| return 0 | ||
|
|
||
| batch_request_id = uuid.uuid4().hex | ||
| ph = ",".join("?" * len(ids_to_delete)) | ||
|
|
||
| # Emit hard_delete events BEFORE the DELETE, in the same transaction. | ||
| for eid in ids_to_delete: | ||
| _append_event_stmt( | ||
| self.conn, | ||
| org_id=self.org_id, | ||
| entity_type=entity_type, | ||
| entity_id=eid, | ||
| op="hard_delete", | ||
| prov="wasInvalidatedBy", | ||
| source_ids=[], | ||
| actor="system", | ||
| request_id=batch_request_id, | ||
| reason="ttl-gc", | ||
| ) | ||
|
|
||
| # Inline FTS/vec cleanup — raw DELETE to preserve atomicity. | ||
| # Do NOT call self._fts_delete/_vec_delete: they self-commit. | ||
| if entity_type in ("user_playbook", "agent_playbook"): | ||
| kind = "user" if entity_type == "user_playbook" else "agent" | ||
| int_ids = [int(eid) for eid in ids_to_delete] | ||
| int_ph = ",".join("?" * len(int_ids)) | ||
| self.conn.execute( | ||
| f"DELETE FROM {kind}_playbooks_fts WHERE rowid IN ({int_ph})", | ||
| int_ids, | ||
| ) | ||
| if self._has_sqlite_vec: # type: ignore[attr-defined] | ||
| self.conn.execute( | ||
| f"DELETE FROM {kind}_playbooks_vec WHERE rowid IN ({int_ph})", | ||
| int_ids, | ||
| ) | ||
| else: | ||
| # profiles: FTS keyed on TEXT profile_id; vec keyed on implicit rowid. | ||
| self.conn.execute( | ||
| f"DELETE FROM profiles_fts WHERE profile_id IN ({ph})", | ||
| ids_to_delete, | ||
| ) | ||
| if self._has_sqlite_vec: # type: ignore[attr-defined] | ||
| rowid_rows = self.conn.execute( | ||
| f"SELECT rowid FROM profiles WHERE profile_id IN ({ph})", # noqa: S608 | ||
| ids_to_delete, | ||
| ).fetchall() | ||
| if rowid_rows: | ||
| rowids = [r[0] for r in rowid_rows] | ||
| rowid_ph = ",".join("?" * len(rowids)) | ||
| self.conn.execute( | ||
| f"DELETE FROM profiles_vec WHERE rowid IN ({rowid_ph})", | ||
| rowids, | ||
| ) | ||
|
|
||
| cur = self.conn.execute( | ||
| f"DELETE FROM {table} WHERE {pk} IN ({ph})", # noqa: S608 | ||
| ids_to_delete, | ||
| ) | ||
| self.conn.commit() | ||
|
|
There was a problem hiding this comment.
Rollback on failure to keep GC truly atomic.
This block performs multiple writes before one final commit. If any step raises, there’s no rollback, so partial writes (notably hard_delete events) can be left pending and later committed by unrelated operations.
Proposed fix
with self._lock:
- rows = self.conn.execute(select_sql, select_params).fetchall()
- if not rows:
- return 0
+ try:
+ rows = self.conn.execute(select_sql, select_params).fetchall()
+ if not rows:
+ return 0
- candidate_ids: list[str] = [str(r[0]) for r in rows]
- ids_to_delete: list[str] = []
- for eid in candidate_ids:
- if self._is_on_legal_hold(self.org_id, entity_type, eid):
- capture_anomaly(
- "lineage.gc.legal_hold_skip",
- level="info",
- org_id=self.org_id,
- entity_type=entity_type,
- entity_id=eid,
- )
- continue
- ids_to_delete.append(eid)
+ candidate_ids: list[str] = [str(r[0]) for r in rows]
+ ids_to_delete: list[str] = []
+ for eid in candidate_ids:
+ if self._is_on_legal_hold(self.org_id, entity_type, eid):
+ capture_anomaly(
+ "lineage.gc.legal_hold_skip",
+ level="info",
+ org_id=self.org_id,
+ entity_type=entity_type,
+ entity_id=eid,
+ )
+ continue
+ ids_to_delete.append(eid)
- if not ids_to_delete:
- return 0
+ if not ids_to_delete:
+ return 0
- batch_request_id = uuid.uuid4().hex
- ph = ",".join("?" * len(ids_to_delete))
+ batch_request_id = uuid.uuid4().hex
+ ph = ",".join("?" * len(ids_to_delete))
- # Emit hard_delete events BEFORE the DELETE, in the same transaction.
- for eid in ids_to_delete:
- _append_event_stmt(
- self.conn,
- org_id=self.org_id,
- entity_type=entity_type,
- entity_id=eid,
- op="hard_delete",
- prov="wasInvalidatedBy",
- source_ids=[],
- actor="system",
- request_id=batch_request_id,
- reason="ttl-gc",
- )
+ # existing mutation logic unchanged...
- ...
- cur = self.conn.execute(
- f"DELETE FROM {table} WHERE {pk} IN ({ph})", # noqa: S608
- ids_to_delete,
- )
- self.conn.commit()
+ cur = self.conn.execute(
+ f"DELETE FROM {table} WHERE {pk} IN ({ph})", # noqa: S608
+ ids_to_delete,
+ )
+ self.conn.commit()
+ except Exception:
+ self.conn.rollback()
+ raise📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| with self._lock: | |
| rows = self.conn.execute(select_sql, select_params).fetchall() | |
| if not rows: | |
| return 0 | |
| candidate_ids: list[str] = [str(r[0]) for r in rows] | |
| ids_to_delete: list[str] = [] | |
| for eid in candidate_ids: | |
| if self._is_on_legal_hold(self.org_id, entity_type, eid): | |
| capture_anomaly( | |
| "lineage.gc.legal_hold_skip", | |
| level="info", | |
| org_id=self.org_id, | |
| entity_type=entity_type, | |
| entity_id=eid, | |
| ) | |
| continue | |
| ids_to_delete.append(eid) | |
| if not ids_to_delete: | |
| return 0 | |
| batch_request_id = uuid.uuid4().hex | |
| ph = ",".join("?" * len(ids_to_delete)) | |
| # Emit hard_delete events BEFORE the DELETE, in the same transaction. | |
| for eid in ids_to_delete: | |
| _append_event_stmt( | |
| self.conn, | |
| org_id=self.org_id, | |
| entity_type=entity_type, | |
| entity_id=eid, | |
| op="hard_delete", | |
| prov="wasInvalidatedBy", | |
| source_ids=[], | |
| actor="system", | |
| request_id=batch_request_id, | |
| reason="ttl-gc", | |
| ) | |
| # Inline FTS/vec cleanup — raw DELETE to preserve atomicity. | |
| # Do NOT call self._fts_delete/_vec_delete: they self-commit. | |
| if entity_type in ("user_playbook", "agent_playbook"): | |
| kind = "user" if entity_type == "user_playbook" else "agent" | |
| int_ids = [int(eid) for eid in ids_to_delete] | |
| int_ph = ",".join("?" * len(int_ids)) | |
| self.conn.execute( | |
| f"DELETE FROM {kind}_playbooks_fts WHERE rowid IN ({int_ph})", | |
| int_ids, | |
| ) | |
| if self._has_sqlite_vec: # type: ignore[attr-defined] | |
| self.conn.execute( | |
| f"DELETE FROM {kind}_playbooks_vec WHERE rowid IN ({int_ph})", | |
| int_ids, | |
| ) | |
| else: | |
| # profiles: FTS keyed on TEXT profile_id; vec keyed on implicit rowid. | |
| self.conn.execute( | |
| f"DELETE FROM profiles_fts WHERE profile_id IN ({ph})", | |
| ids_to_delete, | |
| ) | |
| if self._has_sqlite_vec: # type: ignore[attr-defined] | |
| rowid_rows = self.conn.execute( | |
| f"SELECT rowid FROM profiles WHERE profile_id IN ({ph})", # noqa: S608 | |
| ids_to_delete, | |
| ).fetchall() | |
| if rowid_rows: | |
| rowids = [r[0] for r in rowid_rows] | |
| rowid_ph = ",".join("?" * len(rowids)) | |
| self.conn.execute( | |
| f"DELETE FROM profiles_vec WHERE rowid IN ({rowid_ph})", | |
| rowids, | |
| ) | |
| cur = self.conn.execute( | |
| f"DELETE FROM {table} WHERE {pk} IN ({ph})", # noqa: S608 | |
| ids_to_delete, | |
| ) | |
| self.conn.commit() | |
| with self._lock: | |
| try: | |
| rows = self.conn.execute(select_sql, select_params).fetchall() | |
| if not rows: | |
| return 0 | |
| candidate_ids: list[str] = [str(r[0]) for r in rows] | |
| ids_to_delete: list[str] = [] | |
| for eid in candidate_ids: | |
| if self._is_on_legal_hold(self.org_id, entity_type, eid): | |
| capture_anomaly( | |
| "lineage.gc.legal_hold_skip", | |
| level="info", | |
| org_id=self.org_id, | |
| entity_type=entity_type, | |
| entity_id=eid, | |
| ) | |
| continue | |
| ids_to_delete.append(eid) | |
| if not ids_to_delete: | |
| return 0 | |
| batch_request_id = uuid.uuid4().hex | |
| ph = ",".join("?" * len(ids_to_delete)) | |
| # Emit hard_delete events BEFORE the DELETE, in the same transaction. | |
| for eid in ids_to_delete: | |
| _append_event_stmt( | |
| self.conn, | |
| org_id=self.org_id, | |
| entity_type=entity_type, | |
| entity_id=eid, | |
| op="hard_delete", | |
| prov="wasInvalidatedBy", | |
| source_ids=[], | |
| actor="system", | |
| request_id=batch_request_id, | |
| reason="ttl-gc", | |
| ) | |
| # Inline FTS/vec cleanup — raw DELETE to preserve atomicity. | |
| # Do NOT call self._fts_delete/_vec_delete: they self-commit. | |
| if entity_type in ("user_playbook", "agent_playbook"): | |
| kind = "user" if entity_type == "user_playbook" else "agent" | |
| int_ids = [int(eid) for eid in ids_to_delete] | |
| int_ph = ",".join("?" * len(int_ids)) | |
| self.conn.execute( | |
| f"DELETE FROM {kind}_playbooks_fts WHERE rowid IN ({int_ph})", | |
| int_ids, | |
| ) | |
| if self._has_sqlite_vec: # type: ignore[attr-defined] | |
| self.conn.execute( | |
| f"DELETE FROM {kind}_playbooks_vec WHERE rowid IN ({int_ph})", | |
| int_ids, | |
| ) | |
| else: | |
| # profiles: FTS keyed on TEXT profile_id; vec keyed on implicit rowid. | |
| self.conn.execute( | |
| f"DELETE FROM profiles_fts WHERE profile_id IN ({ph})", | |
| ids_to_delete, | |
| ) | |
| if self._has_sqlite_vec: # type: ignore[attr-defined] | |
| rowid_rows = self.conn.execute( | |
| f"SELECT rowid FROM profiles WHERE profile_id IN ({ph})", # noqa: S608 | |
| ids_to_delete, | |
| ).fetchall() | |
| if rowid_rows: | |
| rowids = [r[0] for r in rowid_rows] | |
| rowid_ph = ",".join("?" * len(rowids)) | |
| self.conn.execute( | |
| f"DELETE FROM profiles_vec WHERE rowid IN ({rowid_ph})", | |
| rowids, | |
| ) | |
| cur = self.conn.execute( | |
| f"DELETE FROM {table} WHERE {pk} IN ({ph})", # noqa: S608 | |
| ids_to_delete, | |
| ) | |
| self.conn.commit() | |
| except Exception: | |
| self.conn.rollback() | |
| raise |
🧰 Tools
🪛 OpenGrep (1.22.0)
[ERROR] 437-440: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
[ERROR] 442-445: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
[ERROR] 448-451: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
[ERROR] 453-456: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
[ERROR] 460-463: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
[ERROR] 465-468: SQL query built via f-string passed to execute()/executemany(). Use parameterized queries with placeholders instead.
(coderabbit.sql-injection.python-fstring-execute)
🪛 Ruff (0.15.17)
[error] 438-438: Possible SQL injection vector through string-based query construction
(S608)
[error] 443-443: Possible SQL injection vector through string-based query construction
(S608)
[error] 449-449: Possible SQL injection vector through string-based query construction
(S608)
[error] 461-461: Possible SQL injection vector through string-based query construction
(S608)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@reflexio/server/services/storage/sqlite_storage/_lineage.py` around lines 391
- 470, Wrap the entire operation block (from the _append_event_stmt call through
the self.conn.execute DELETE statement and self.conn.commit) in a try-except
block to ensure atomicity. In the except clause, call self.conn.rollback() to
undo any partial writes if an exception occurs at any step, then re-raise the
exception. This prevents partial writes from persisting when unrelated
operations commit later.
Collaborator
Author
|
CodeRabbit findings addressed in follow-up PR #193 (commit |
yilu331
added a commit
that referenced
this pull request
Jun 21, 2026
/#191) (#193) ## What Addresses the CodeRabbit review findings left on the **merged** lineage PRs **#187 (B1)**, **#188 (B2)**, and **#191 (B3-pre)**. Pure remediation — no new features. ## Fixes by source PR **#191 (B3-pre)** — `22d2540`: negative-`limit` guard (`<=0`) in reconstruction; `enabled_org_ids or []` defensiveness in feature flags; parity script **fail-closed on duplicate `request_id`** + **INCONCLUSIVE (exit 2) on at-cap/truncated reads**; test hygiene (vacuous-MATCH guard, RECON-vs-LEGACY label, exact request_id set-equality, empty-id seeding); ASCII `union` (RUF002). **#188 (B2)** — `4204598`: `LineageGCConfig` bounds (`Field(gt=0)`); scheduler poll-interval clamp; `gc_expired_tombstones` `limit<=0` guard; **explicit rollback** to keep GC atomic on mid-write failure. **#187 (B1)** — `a5f6558` + `f792ddb`: **phantom-audit guards on the bulk-delete paths** (emit `hard_delete` only for rows that exist, in the **same commit** as the base DELETE; FTS/vec cleanup moved **after** the commit per the SQLite self-commit rule) — fixing `delete_all_agent_playbooks` and `delete_archived_agent_playbooks_by_playbook_name`, which emitted *before* the mutation; `entity_type` filter in a lineage-lookup test. ## Deliberately NOT changed (with reasons) - **SQLite `get_profiles_by_generated_from_request_id` "not org-scoped"** — SQLite `profiles` has **no `org_id` column** (tenant isolation is per-DB-file; enterprise is per-schema), and the reconstruction's event pool is already `org_id`-scoped via `get_lineage_events`. Adding a profiles-level org filter needs a schema migration — tracked as a separate follow-up, not a quick fix. - **`_set_config(**kwargs)` `🔴 Critical`** — **confirmed false alarm**: `Config.model_validate({..., **overrides})` is valid Pydantic; the test passes and applies the right config. - Several #187 items were **already addressed** by later phases (rowcount gates, already-archived exclusion, atomicity, existing-row filters, vec-sidecar cleanup, `request_id` non-optional, `StorageError` wrapping) — verified against current code and skipped. All changes tested (reviewed via a final whole-branch pass that caught the bulk-delete ordering). No behavior change beyond the hardening above. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added validation for scheduler configuration values to ensure they meet minimum requirements. * **Bug Fixes** * Improved transaction atomicity during garbage collection operations. * Fixed feature flag evaluation to safely handle missing or null org ID lists. * Enhanced duplicate detection in data parity validation. * Enforced minimum scheduler poll interval to improve reliability. * **Tests** * Added comprehensive tests for configuration validation and garbage collection scenarios. * Strengthened parity checking and deletion operation test coverage. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This was referenced Jun 21, 2026
This was referenced Jun 30, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Lineage Phase B2 — TTL GC (tombstone garbage collection), OSS/SQLite half. Ships OFF by default.
The stage-2 of the soft-delete→hard-purge model: a scheduled, per-org, config-gated job that physically hard-deletes aged tombstones (status ∈ {merged, superseded, archived}, older than a retention window), emitting one content-free
hard_deletelineage event per deleted row first — reusing the B1 emit-before-delete atomic pattern.Changes
LineageGCConfig(config_schema.py) —enabled=False,tombstone_grace_window_days=90,poll_interval_seconds=86400. Off by default; attached to rootConfig.gc_expired_tombstones(*, entity_type, older_than_epoch, limit)(storage base + SQLite) — type-correct age comparison (TEXT-ISOcreated_atvia_epoch_to_isofor playbooks, INTEGERlast_modified_timestampfor profiles), exclusive cutoff, eligible-status set{merged,superseded,archived}(its own set, not_TOMBSTONE), emit-then-delete in one lock/one commit with raw inline FTS/vec cleanup (no self-committing helpers), real per-call batchrequest_id, idempotent. Plus a_is_on_legal_holddeferred seam.LineageGCScheduler+ lifespan wiring (gc_scheduler.py,api.py) — daemon mirroringresume_scheduler; per-org config gate;capture_anomalycheckpoints (lineage.gc.run_failed/high_volume/legal_hold_skip); no bookmark (GC is idempotent); mid-tick stop.list_org_idsdeclared on the base (SQLite returns[self.org_id]); the scheduler degrades to bootstrap-org with a logged warning where unimplemented.Safety
GC is triple-gated off (config default, scheduler factory, per-org gate) and cannot be enabled until the documented pre-enable gates are resolved (tuner-window pinning PB-9, B2↔B3 timing PB-5, the age-basis decision PB-8b, a real legal-hold store, enterprise
list_org_ids). This PR builds the mechanism only.Built via subagent-driven-development: per-task implement → review → fix → re-review, plus a final whole-branch review.
Summary by CodeRabbit
New Features
Improvements