Ingestion pipeline reference

Supported inputs#

Every runtime of the adapter registry (src/adapters/mod.rs) is a supported input; transcript-lake sources --json lists each with the roots it reads on the machine it runs on, and each adapter's module documentation names its store path, every record shape it reads and the upstream source the shapes come from. Stores fall into two readings:

A directory that disappears during discovery yields no entries; an entry that exists but cannot be read, or whose type cannot be read, is named on stderr with its cause (sessions: cannot list <dir>: <cause>) and the sessions beside it are still read. Unowned paths and unrelated artifacts are ignored. A complete line that is not JSON emits no event and is named on stderr: sessions: the line at byte <offset> of <file> is not JSON and carries no record: <parser error>. A partial final line waits for a later append, and complete prior lines still stream.

Devin ingestion excludes hidden sessions and nodes explicitly marked is_system_prefix. Other recorded system messages, including injected hook context, retain their masked text as meta events with extra.kind = system_prompt; they are not discarded with the rebuilt prefix scaffolding.

Devin's node table is not the conversation: every node names its parent, the agent rebuilds its context chain whenever its system prefix changes (so several nodes repeat one message_id), a subagent runs its own chain declared by a subagent_heads row, and a rewind leaves a chain behind that no head names. Ingestion rebuilds each session's chains from that graph. The live conversation is the chain sessions.main_chain_id heads, read from its root in conversation order so a repeated message is kept once at its first copy (the user turn opens the chain, not a later compacted replay of it), and its events carry the session's own id. Each declared subagent chain is its own session, <session>:<agent_id>, and each chain no head names is <session>:branch-<root node>, named by its root-most node no other chain covered; their events carry extra.parent_session and extra.chain. A node a chain already covered, or a message a chain already kept, is never emitted again as branch content, so a branch that shares the live chain's prefix emits only what the rewind discarded. A main_chain_id or chain_node_id naming no stored node, a parent link naming no node, and a parent link that returns to a node already walked each refuse the adoption naming the session and the node, so a store contradicting itself is not read as a shorter conversation. The repository's tests/devin/chains.rs journey writes the observed-format registry's Devin fixture into a real store under $XDG_DATA_HOME/devin/cli, adds a rewound branch rooted on the live chain, runs transcript-lake adopt over it and reads the persisted partitions back (cargo test --test devin-chains).

A disconnected parent cycle with no declared head and no leaf also refuses ingestion instead of silently omitting its messages. The diagnostic includes the database path, session and an uncovered node: node <id> is outside every headed or leaf chain: its parent links form a cycle. The native store remains read-only; this refusal does not repair or delete its nodes.

Devin refuses a node whose message or non-null metadata is not valid JSON. The diagnostic names the database path, session, node, failed field and parser cause. Unreadable metadata cannot classify a system prefix as ordinary conversation, and an unreadable message cannot become a successful shortened conversation. The whole-store reader returns no events from that load; the vendor database remains read-only. Correct the source through its owning application before adopting again. The real CLI refusal case is cargo test --test devin-damaged-node; a test command listed here is not a runtime qualification result.

Cursor likewise refuses malformed non-null JSON in a composer or bubble record. Its error retains the record key, state.vscdb path and parser cause; SQL NULL remains an absent value. No shortened conversation is returned from that whole-store load. The isolated native-store journey cargo test --test cursor-damaged-record covers refusal before event publication, repair of the owned source, recovery of every message occurrence and unchanged persisted events on replay.

The other SQLite stores read whole refuse a damaged stored value the same way, through the one shared reader of stored JSON: an amazon-q or kiro-cli conversation (named by its directory), a crush message's parts (named by role, session and creation second), a goose message's content (named by role and session), and an opencode, kilo or zcode part or the message it belongs to (named by session). The error reads <record> in <database> is not JSON: <parser cause>, and that load returns no events rather than the rest of a conversation with the damaged record missing.

Native-store qualification

The repository's tests/harnesses/ingestion.rs journey uses existing native stores, not generated transcript fixtures. Prepare a JSON array of independently recorded expectations; each object supplies runtime, the discovered root, session_id, and exact masked user_text. Use stable completed sessions: ongoing vendor writes can legitimately change a partition between the two adoption calls.

LAKE_TEST_EXPECTATIONS=/absolute/path/to/native-store-expectations.json \
LAKE_TEST_MIN_HARNESSES=40 \
cargo test --test harness-ingestion native_stores_preserve_messages_and_replay_without_duplicates -- --exact --ignored --nocapture

Run from the product repository when execution is permitted. The required distinct-harness count is explicit and retained with the evidence; a smaller count does not qualify the at-least-40-harness contract. Each case adopts a real discovered store into its own build-directory Lake, checks a known user message in persisted canonical partitions, repeats adoption and compares partition bytes to detect duplicates, then checks that an undiscovered root refuses without writing events or selecting a source. Missing stores or messages fail rather than becoming skips. The journey retains source revision, source patch, binary digest, exact expectations, commands, exit statuses, output and resulting partitions under the Cargo build directory. Creating this test source is not evidence that these real flows have passed.

Successful adoption must also persist sources.json with one declaration matching its selected source identity, runtime and canonical native root. The journey reads that persisted state rather than trusting the command response, and repeated adoption must preserve the complete selection document. Imported messages without a durable source selection do not qualify.

A journey's retained record says whether each command started, is still running, exited or failed to start, beside its output. An interrupted running record is incomplete evidence, not a passing result: read its partial output instead of repeating the adoption to recover diagnostics.

For Devin graph qualification, select devin_native_data_directory_is_discovered_and_adopted with --exact --ignored --nocapture on the same harness-ingestion target. Set LAKE_TEST_DEVIN_DATA_HOME to the existing native data directory and LAKE_TEST_DEVIN_EXPECTATION to an independently recorded JSON object containing runtime, root, session_id, user_text, context_text and chains. The root must be that data directory's devin/cli directory with its real sessions.db.

The chains object supplies main, an array of every expected canonical message's event_type and masked text, plus subagent and branch objects. Each child object supplies its independently observed session_id, chain identity and complete messages array in the same format. Use a completed native session with both a subagent and discarded rewind content. The journey compares exact message multisets, including duplicate counts, checks child parent/chain metadata, and refuses missing expectations. Flattening child messages into the main session, losing branch content or retaining rebuilt duplicate messages cannot satisfy these expectations. This does not assert message ordering or replace the separate forty-harness qualification.

The same chains object must contain a nonempty hidden_sessions array of independently observed hidden session IDs from that native store. Choose hidden sessions that actually contain recorded messages; nonexistent IDs do not qualify. The journey checks that neither those sessions nor their child chains appear in persisted events, by session identity and parent metadata. A store without such a native case cannot qualify hidden-session exclusion.

Before adoption, the journey opens the native database read-only and verifies each excluded ID against sessions.hidden joined to its recorded message_nodes. Missing IDs, visible sessions and sessions without message nodes fail the prerequisite check; an arbitrary nonexistent ID cannot make the exclusion assertion pass.

Raw-to-canonical transformation#

An adapter returns RawEvent { ts, session_id, project, event_type, text, tool_name, model, tokens_in, tokens_out, extra }. The stream adds stable runtime and local machine, masks every retained string, caps values, then serializes this exact canonical order:

1. ts: string|null
2. runtime: string
3. machine: string
4. session_id: string|null
5. project: string|null
6. event_type: string
7. text: string|null (null when the runtime recorded no text)
8. tool_name: string|null
9. model: string|null
10. tokens_in: integer|null
11. tokens_out: integer|null
12. extra: object

Allowed types are user, assistant, thinking, tool_call, tool_result, meta, and hook_decision. extra.source_stem_hash is SHA-256 of the source filename stem; the plaintext stem is not stored.

Catch-up and live append#

1. Create the platform watcher (FSEvents on macOS, inotify on Linux through notify::recommended_watcher).
2. Recursively register the selected source root, or all discovered roots before a source is selected.
3. Run startup catch-up by enumerating candidate files.
4. Skip a file only when its saved source identity, change timestamp, mtime, size, and complete offset all match.
5. For a modified file, hash its consumed prefix once. Resume only if the digest matches; otherwise replay that source from the beginning against its retained canonical occurrences.
6. Consume complete newline-terminated records. A partial final line waits for a later append.
7. Coalesce queued filesystem notifications and resolve only changed paths through entry_for.
8. On SIGINT/SIGTERM, unwatch roots, publish final status, and release the lease.

The loop blocks until a watched path changes or SIGINT/SIGTERM arrives through its own stop channel; it does not wake on a timer. Source reads go through a standard buffered reader with no buffer size chosen by the Lake; a batch is the events parsed from what one read handed over, and it is written and checkpointed when that read is used up, with no event count chosen by the Lake. Ordinary append batches checkpoint after their outputs succeed. Recovery keeps its pre-replay cursor until all complete source lines succeed, so an interrupted replay remains recoverable without duplicated suffixes.

Commit protocol#

For each source delta:

1. Claim the single writer lease.
2. Verify the consumed source prefix, or load retained occurrence counts for the affected source's deterministic partition files.
3. Parse complete source lines, mask/cap events, and group new canonical occurrences by date.
4. Append and fsync complete payloads to the daily partitions.
5. Update affected Oko session projections, including previously archived rows that a projection lacks.
6. Record the complete-line offset, consumed-prefix SHA-256, and opened-file identity through atomic, fsynced cursor publication.
7. Emit per-runtime event/mask counts and release the lease.

Changed prefixes, shorter sources, and legacy cursors recover automatically. Diagnostics name the source, previous offset, observed bytes, and one reason: legacy cursor has no verified source prefix, source is shorter than its consumed prefix, or source bytes before the cursor changed. Summaries expose replayedFiles per runtime; exact-source ingestion also returns filesReplayed. Recovery preserves masked canonical history and counts retained occurrences instead of appending them again. If interrupted, the old cursor remains and the next attempt repeats that comparison. Corrupt canonical input refuses with cannot replay source against corrupt canonical partition <path> at line <line>: <cause>. A partition write failure names cannot commit canonical partition <path>: <cause>. Neither refusal advances the recovery cursor.

Hook closed segments#

Closed segments are immutable producer publications. The stream content-checks each output. A pre-existing output with a different digest refuses with hook segment output conflict: <path>. Successful output uses temp write, file fsync, rename, and parent fsync; committed segment records make replay idempotent. Legacy mutable hook logs are used only when the closed-segment ready directory is absent.

Replay recovery#

rebuild --to <empty-path> [--source <runtime>] obtains a distinct empty destination, enumerates selected source history, and runs the same parse/mask/canonicalize path. It refuses the current root or non-empty destination. It never repairs in place or deletes the preserved Lake.