feat(tapes): Cognee as a Paper/tapes cassette (cassette/v1alpha1) - #384
feat(tapes): Cognee as a Paper/tapes cassette (cassette/v1alpha1)#384Vasilije1990 wants to merge 2 commits into
Conversation
Implements cognee as a first-class tapes cassette: a FastAPI service that tapes discovers via --cassettes, validates against the x-tapes-cassette manifest served at /openapi, and proxies under /v1/cassettes/cognee/. - GET /ping + GET /openapi (hand-authored contract, conformance-tested against the actually served routes) - POST /api/sync, /api/sync/status, /api/search — exposed as MCP tools cognee.sync_sessions / cognee.sync_status / cognee.search_memory - Incremental sync checkpoints from /v1/sessions list items' last_seen_at (confirmed field), sidestepping the unverified export-payload location flagged in PR #362; only completed sessions advance the checkpoint - Content-hash idempotency, pending_cognify recovery, per-session failure tolerance, optional COGNEE_STORAGE_ROOT isolation - 20 offline tests (tapes mocked at httpx transport, cognee stubbed) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| async def sync(request: SyncRequest | None = None) -> dict: | ||
| request = request or SyncRequest() | ||
| if request.wait: | ||
| if syncer.is_running(): | ||
| return {"accepted": False, "status": syncer.status.snapshot()} | ||
| status = await syncer.run(full=request.full) | ||
| return {"accepted": True, "status": status.snapshot()} |
There was a problem hiding this comment.
blocker — Race condition breaks single-flight guarantee.
When wait=True, this path checks is_running() then calls await syncer.run() directly (outside the task system). Between the check and the run() call, another request with wait=False could call syncer.start(), creating a background task. Now two syncs run concurrently: one in the task (from start()), one directly (from this path). Both will read/write the same state file, causing races.
Fix: Make wait=True also use task-based sync, just with await self._task:
if request.wait:
if not syncer.start(full=request.full):
return {"accepted": False, "status": syncer.status.snapshot()}
status = await syncer._task # wait for the task we just started
return {"accepted": True, "status": status}
|
|
||
| try: | ||
| export = await self._tapes.export_session(session_id) | ||
| except Exception as exc: # noqa: BLE001 — skip and continue with the rest |
There was a problem hiding this comment.
major — Bare except Exception catches asyncio.CancelledError, preventing clean task cancellation.
If the app shuts down or the task is cancelled while exporting, this will catch the CancelledError and continue looping instead of propagating the cancellation.
Fix: Exclude cancellation from the catch:
except (httpx.HTTPError, ValueError, KeyError, json.JSONDecodeError) as exc:
| self._save_state(state) | ||
|
|
||
| status.state = "completed" | ||
| except Exception as exc: # noqa: BLE001 — surfaced via status, not a crashed task |
There was a problem hiding this comment.
major — Same issue: bare except Exception catches asyncio.CancelledError.
If the background task is cancelled (e.g., server shutdown), this will catch the cancellation and surface it as a failed sync instead of propagating it.
Fix: Narrow the exception or re-raise CancelledError:
except asyncio.CancelledError:
raise # propagate cancellation
except Exception as exc:
| return _State() | ||
|
|
||
| def _save_state(self, state: _State) -> None: | ||
| self._config.state_path.write_text(json.dumps(asdict(state), indent=2)) |
There was a problem hiding this comment.
major — Non-atomic write risks state file corruption.
write_text() isn't atomic: if the process crashes mid-write, the JSON is corrupted. You handle reading corrupted files (lines 84-85), but that loses all progress since the last good write.
Fix: Write-to-temp-then-rename for atomicity:
import tempfile
tmp = self._config.state_path.with_suffix('.tmp')
tmp.write_text(json.dumps(asdict(state), indent=2))
tmp.replace(self._config.state_path) # atomic on POSIX
| @asynccontextmanager | ||
| async def lifespan(_: FastAPI): | ||
| yield | ||
| await tapes.aclose() |
There was a problem hiding this comment.
minor — Missing graceful sync task cancellation on shutdown.
If the server shuts down while a sync is running, the lifespan only closes the HTTP client but doesn't wait for syncer._task. The background task gets cancelled abruptly, potentially leaving pending_cognify=True but cognify not actually run.
Fix: Cancel and await the task:
@asynccontextmanager
async def lifespan(_: FastAPI):
yield
if syncer._task and not syncer._task.done():
syncer._task.cancel()
try:
await syncer._task
except asyncio.CancelledError:
pass
await tapes.aclose()
| items: list[dict] = [] | ||
| cursor: str | None = None | ||
| while True: | ||
| params: dict = {"limit": 100} | ||
| if cursor: | ||
| params["cursor"] = cursor | ||
| response = await self._client.get(f"{self._base_url}/v1/sessions", params=params) | ||
| response.raise_for_status() | ||
| payload = response.json() | ||
| items.extend(item for item in payload.get("items", []) if isinstance(item, dict)) | ||
| cursor = payload.get("next_cursor") | ||
| if not cursor: | ||
| break |
There was a problem hiding this comment.
minor — Unbounded pagination could exhaust memory.
A full=True sync with millions of sessions will load all list items into memory. The commit message mentions incremental checkpoints help, but there's no safety limit if tapes has massive history.
Consider: Add a max items limit or stream processing instead of accumulating all items:
# Option 1: Hard limit
if len(items) > MAX_SESSIONS:
logger.warning("Hit session list limit, results truncated")
break
# Option 2: Yield items as found (requires refactoring caller)
|
20 tests, 139-line README, and a commit message that reads like a design doc — yet the wait=True path can still double-book the sync and corrupt the state file. 🎭 🔴 1 blocker, 3 major — changes requested
See inline comments for details. |
CI runs `ruff format --check integrations/`; three files in the new cassette were unformatted. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Elegant API surface, comprehensive tests, and a state-persistence layer that corrupts itself on crash — the sync machinery reads like production but the atomicity reads like a prototype. 🔴 1 blocker, 3 major — changes requested
See inline comments for details. |
| return _State() | ||
|
|
||
| def _save_state(self, state: _State) -> None: | ||
| self._config.state_path.write_text(json.dumps(asdict(state), indent=2)) |
There was a problem hiding this comment.
blocker — State file write is not atomic. If the process crashes or is killed mid-write (e.g., during the sync loop at line 153), the state file will be corrupted with partial JSON, and the next run will log "starting fresh" and lose all checkpoint/hash state.
Fix: Use atomic write pattern:
def _save_state(self, state: _State) -> None:
import tempfile
content = json.dumps(asdict(state), indent=2)
with tempfile.NamedTemporaryFile(
mode='w', dir=self._config.state_path.parent, delete=False
) as tmp:
tmp.write(content)
tmp.flush()
os.fsync(tmp.fileno())
Path(tmp.name).replace(self._config.state_path)Atomicity matters here because state writes happen inside the sync loop (line 153) where the process is vulnerable to SIGKILL/OOM/crashes.
| text = build_transcript(export) | ||
| if not text: | ||
| status.skipped += 1 | ||
| continue | ||
|
|
||
| text_hash = hashlib.sha256(text.encode("utf-8")).hexdigest() | ||
| if state.sessions.get(session_id) == text_hash: | ||
| status.unchanged += 1 | ||
| else: | ||
| logger.info("Ingesting session %s.", session_id) | ||
| await cognee.add(data=text, dataset_name=self._config.dataset_name) | ||
| state.sessions[session_id] = text_hash | ||
| state.pending_cognify = True | ||
| status.ingested += 1 | ||
| self._save_state(state) # incremental — protects progress mid-run | ||
|
|
||
| if last_seen_at := item.get("last_seen_at"): | ||
| try: | ||
| seen = parse_ts(last_seen_at) | ||
| except ValueError: | ||
| continue | ||
| if latest_completed is None or seen > latest_completed: | ||
| latest_completed = seen |
There was a problem hiding this comment.
major — Infinite retry loop for legitimately empty transcripts. If a completed session produces an empty transcript (e.g., session has only thinking blocks, no main LLM spans, or malformed trace structure), it:
- Gets skipped at line 141 (correctly — nothing to ingest)
- Never reaches the
latest_completedupdate at lines 155-161 - Checkpoint never advances past it
- Gets fetched and retried on every subsequent sync forever
This wastes API calls and sync time. Sessions with export failures are retried (correct), but sessions with structural emptiness (e.g., build_transcript returns "" due to missing main spans) should be recorded as processed-but-empty so they don't block progress.
Fix: Track empty-transcript sessions in state:
if not text:
# Mark as processed-but-empty so we don't retry forever
state.sessions[session_id] = "EMPTY_TRANSCRIPT"
self._save_state(state)
status.skipped += 1
# Still update checkpoint for this session
else:
text_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()
if state.sessions.get(session_id) == text_hash:
status.unchanged += 1
else:
logger.info("Ingesting session %s.", session_id)
await cognee.add(data=text, dataset_name=self._config.dataset_name)
state.sessions[session_id] = text_hash
state.pending_cognify = True
status.ingested += 1
self._save_state(state)
if last_seen_at := item.get("last_seen_at"):
# (this block stays the same)Then the checkpoint advances past empty sessions and they're not retried.
| if state.pending_cognify: | ||
| await cognee.cognify(datasets=[self._config.dataset_name]) | ||
| state.pending_cognify = False | ||
| self._save_state(state) | ||
| logger.info("Cognify run complete.") |
There was a problem hiding this comment.
major — Cognify failure leaves ingested sessions orphaned. If cognify() fails after sessions were added (line 149) and saved to state (line 153), the exception is swallowed by the outer except at line 174, pending_cognify gets reset to False at line 165, and those sessions are never cognified. They're in the raw dataset but not in the graph, and the cassette has no mechanism to retry them.
Fix: Wrap cognify separately and preserve the flag on failure:
if state.pending_cognify:
try:
await cognee.cognify(datasets=[self._config.dataset_name])
state.pending_cognify = False
self._save_state(state)
logger.info("Cognify run complete.")
except Exception as exc:
logger.error("Cognify failed: %s — will retry next sync.", exc)
# Leave pending_cognify=True so next run retries
self._save_state(state)This way, a transient cognify failure (e.g., LLM rate limit, OOM) doesn't lose the sessions — they'll be cognified on the next sync.
| @dataclass | ||
| class _State: | ||
| """On-disk sync state: per-session content hashes + incremental checkpoint.""" | ||
|
|
||
| sessions: dict = field(default_factory=dict) |
There was a problem hiding this comment.
major — Unbounded state growth. The sessions dict accumulates one hash per session forever, never cleaned up. After syncing thousands of sessions over months, the state file (and the _load_state memory footprint) grows indefinitely. A multi-year cassette instance could hit MB-scale state files.
Fix: Add periodic pruning to keep only sessions seen in the last N syncs or days. One approach:
@dataclass
class _State:
sessions: dict = field(default_factory=dict)
last_synced_at: str | None = None
pending_cognify: bool = False
last_pruned_at: str | None = None # track when we last pruned
# Then in run(), after checkpoint update:
if latest_completed is not None:
state.last_synced_at = latest_completed.isoformat()
# Prune old sessions once per day
if self._should_prune(state, latest_completed):
old_count = len(state.sessions)
# Keep only sessions from last 90 days (or last 10k sessions, whichever is larger)
# Implementation left as exercise; basic version: drop entries not in recent items list
state.last_pruned_at = latest_completed.isoformat()
logger.info("Pruned state: %d → %d sessions.", old_count, len(state.sessions))
self._save_state(state)
| return parse_ts(last_seen_at) > parse_ts(since) | ||
| except ValueError: | ||
| # Fall back to string comparison for timestamps we can't parse. | ||
| return last_seen_at >= since |
There was a problem hiding this comment.
minor — Inconsistent comparison operator in fallback path. Main path (line 18) uses > (strictly newer), but the fallback uses >= (newer-or-equal). A session with an unparseable timestamp that lexically equals the checkpoint would be included in the fallback but excluded in the main path. This asymmetry could cause a session to be processed twice (once when the checkpoint timestamp is parseable, again if it later becomes unparseable due to env/TZ changes).
Fix: Use > for consistency:
return last_seen_at > sinceIf exact-match inclusion was intentional, document why and make both paths use >=.
1 participant
What
Implements Cognee memory as a first-class tapes cassette under
integrations/tapes-cassette/, following Paper's cassette-anatomycassette/v1alpha1contract: an independent FastAPI service that tapes discovers via--cassettes localhost:9900/openapi, validates against thex-tapes-cassettemanifest, and proxies under/v1/cassettes/cognee/....It syncs recorded agent sessions into a cognee knowledge graph and exposes memory as MCP tools inside tapes, so agents can query their own session history.
Surface
GET /pingGET /openapix-tapes-cassettemanifestPOST /api/synccognee.sync_sessions{"wait": true}for inline)POST /api/sync/statuscognee.sync_statusPOST /api/searchcognee.search_memoryAll tools are
POSTroutes withx-tapes-mcpannotations, per the v1alpha1 limitation that only POST converts to MCP tools.Relationship to #362 (tapes exporter)
Complementary, not competing: same transcript-extraction rules (completed sessions only, "main" LLM spans, thinking dropped, tool calls summarized to curated keys) so both produce compatible graph content — but this integrates inside the tapes namespace rather than polling from outside.
Notably, it sidesteps the unverified
last_seen_atquestion blocking #362: the incremental checkpoint readslast_seen_atfromGET /v1/sessionslist items (a confirmed field), never from the export payload whose shape was never validated. Only completed sessions advance the checkpoint; an in-progress session's timestamp bumps again on completion, so the next run picks it up.Other hardening over the exporter's known limitations: httpx with connection reuse, per-session export failures skipped without killing the run, a
pending_cognifyflag so an interrupted run finishes its cognify next time, content-hash idempotency, and optionalCOGNEE_STORAGE_ROOTstorage isolation.Testing
uv run pytest -q— 20 passed, fully offline (tapes mocked at the httpx transport layer, cogneeadd/cognify/searchstubbed). Includes manifest-vs-served-routes conformance tests so the hand-authored contract can't drift from the app.uv run ruff check .— clean.cognee-tapes-cassette, verified/ping, manifest correctness (runtime port reflected), and gracefulstate: failedsync status with no tapes server reachable.Also registers the package in
integrations/inventory.yml.Caveats
v1alpha1is alpha — manifest/MCP conventions follow the blog post as published and may need updating as tapes evolves.🤖 Generated with Claude Code