Skip to content
atif-sql

Data flow

View as MarkdownOpen in ChatGPTOpen in ClaudeOpen in Claude CodeOpen in Cursor

atif-sql · Data flow

Section titled “atif-sql · Data flow”

The distribution declares exactly one entry point, the console script atif-sql = "atif_cli.app:main" (packages/atif-cli/pyproject.toml:42), and the main it names installs a WARNING-and-up loguru sink and hands control to cyclopts — packages/atif-cli/src/atif_cli/app.py:1125. Every process below therefore begins as a CLI invocation; there is no HTTP, RPC, or queue surface to enter through.

Two of the three flows below produce the corpus and one consumes it: materialize fills <corpus_root>/sessions/ (:339), analyze fills <corpus_root>/analytics/ (:627), and query binds both and runs caller SQL over them (:497). The other commands are subsets of these three — convert (:225) is one iteration of flow 1’s inner loop, search (:828) re-enters flow 2’s registration at :886 and adds one kNN statement, embed (:734) writes the vector store flows 2 and 3 read, and schema (:1055), examples (:963), and status (:412) answer from static data or stat calls with no downstream participant.

Flow 1: corpus materialization (atif-sql materialize)

Section titled “Flow 1: corpus materialization (atif-sql materialize)”
  1. The materialize command resolves CorpusSettings (pydantic-settings, env prefix ATIF_SQL_), then injects the three things the pure use case will not own: the ConverterPort adapter, the wall-clock instant, and the harbor / converter version pins stamped into every meta.json — packages/atif-cli/src/atif_cli/app.py:352.
  2. One pass through the corpus use case runs scan, plan, convert, and write in that order and returns a MaterializationReport — packages/atif-corpus/src/atif_corpus/application/materialize.py:476.
  3. The scanner discovers every session under the raw transcript root and separates unreadable sessions from absent ones, so a stat failure is never mistaken for a deletion — packages/atif-corpus/src/atif_corpus/infrastructure/scanner.py:143.
  4. The pure planner partitions the scan into to-materialize, up-to-date, and skipped-live using the previous watermark and the quiescence policy; force overrides staleness but never liveness — packages/atif-corpus/src/atif_corpus/domain/sessions.py:159.
  5. For each planned session the use case calls converter.convert across the port; one session’s exception is caught and recorded so a single bad transcript cannot abort the sync — packages/atif-corpus/src/atif_corpus/application/materialize.py:599.
  6. The adapter that satisfies ConverterPort lives in atif-cli because the independence contract forbids atif-corpus from importing atif-converter; it raises rather than returning an invalid trajectory — packages/atif-cli/src/atif_cli/converter_adapter.py:51.
  7. convert_and_audit snapshots the source files, stages the session into harbor’s expected directory shape and calls harbor’s pinned private method (packages/atif-converter/src/atif_converter/infrastructure/harbor_adapter.py:178), then builds the loss report and edges from the raw records, enriches the trajectory, and refuses the result if any source moved mid-pass — packages/atif-converter/src/atif_converter/application/convert_and_audit.py:105.
  8. The four artifacts are written under .staging/ with meta.json last, then the whole directory swaps into sessions/<id>/ so a reader sees one complete generation or the other (packages/atif-corpus/src/atif_corpus/application/materialize.py:240); the watermark advances only for sessions that succeeded — :608.
loop [each planned session] materialize(source_root, corpus_root, ConverterPort) read_watermark + scan_sources SessionSource list, prior mtimes build_plan -> stale / current / live ConverterPort.convert(session_jsonl) pinned private converter(session_dir) ATIF trajectory ConversionOutput stage 4 artifacts, meta.json last, swap dir write watermark.json MaterializationReport atif-cli atif-corpus atif-converter harbor corpus disk

Flow 2: SQL read path (atif-sql query '<sql>')

Section titled “Flow 2: SQL read path (atif-sql query '<sql>')”
  1. The query command resolves the corpus root and reads the active embedder’s (model_id, dim) without constructing the embedder, so the vector store’s stamped identity can be checked before anything binds over it — packages/atif-cli/src/atif_cli/app.py:529.
  2. A DuckDB connection is opened with no path or URI, so the engine runs in-process against memory — :591.
  3. Registration builds the whole catalog on that connection in a fixed order — raw TEMP tables, base views, VSS, macros, analytics views, analytics macros — because each later stage binds against the earlier one at CREATE time — packages/atif-duck/src/atif_duck/infrastructure/registry.py:1212.
  4. The raw readers materialize the four corpus artifact kinds as TEMP TABLEs over globs into <corpus_root>/sessions/, which is where the cost of a query invocation lives: O(corpus) per connection — :196.
  5. Trajectory, edges, and loss readers are semi-joined against the meta table, so a session directory missing its meta.json contributes nothing rather than a partial artifact set. This is the read-side half of flow 1’s meta-last write ordering — packages/atif-duck/src/atif_duck/infrastructure/registry.py:221.
  6. The VSS step ATTACHes the LanceDB store through the lance extension (packages/atif-duck/src/atif_duck/infrastructure/registry.py:887) and reads the store’s stamped model and dim back out (packages/atif-duck/src/atif_duck/infrastructure/registry.py:921) before creating any view over it; a store written by a different provider or width raises instead of binding, because vectors from different models live in incompatible spaces and would return numerically valid but meaningless cosine scores — packages/atif-duck/src/atif_duck/domain/embedding_guard.py:46.
  7. The fully-registered connection is then sandboxed: spill directory, memory cap, a directory allowlist holding only the spill area, a path allowlist holding the individual analytics parquets, enable_external_access=false, and lock_configuration last so caller SQL cannot widen any of it — packages/atif-cli/src/atif_cli/app.py:138.
  8. The caller’s statement executes against the locked connection (:606) and the cursor drains in batches to stdout — a JSON array of row objects on a pipe, a width-aligned table on a TTY — packages/atif-cli/src/atif_cli/output.py:154.
duckdb.connect() register(con, corpus_root, expected model + dim) CREATE TEMP TABLE raw readers over corpus globs read_json sessions/*/meta.json then the rest rows from meta-bearing session dirs only ATTACH lance_store (TYPE LANCE) SELECT model, dim LIMIT 1 stamped identity ensure_store_matches, then bind the view views and macros registered allowlists, external access off, lock_configuration execute(caller SQL) cursor drain in batches to stdout atif-cli atif-duck DuckDB corpus disk Lance store

Flow 3: analytics enrichment (atif-sql analyze --no-dry-run)

Section titled “Flow 3: analytics enrichment (atif-sql analyze --no-dry-run)”
  1. The analyze command loads AnalyticsSettings, reuses the same corpus-root resolution the other commands share, and stamps explicit --max-sessions / --max-cost-usd ceilings over the env defaults so a crontab line carries its spend cap visibly — packages/atif-cli/src/atif_cli/app.py:659.
  2. The pipeline runner builds one shared CorpusReader for every stage, runs the three structural stages, then loops the five LLM stages; the lane selectors and skip_* flags subtract stages from whichever lane runs — packages/atif-analytics/src/atif_analytics/application/analyze.py:36.
  3. Corpus rows are read with stdlib json over <corpus_root>/sessions/<id>/trajectory.json behind a bounded memo — not through DuckDB, which the forbidden import contract puts out of this package’s reach — packages/atif-analytics/src/atif_analytics/infrastructure/corpus_reader.py:269.
  4. The structural lane (cluster, then terms, then community) reads vectors straight out of the Lance store and writes single parquet files, bypassing the sharded cache — packages/atif-analytics/src/atif_analytics/infrastructure/lance_reader.py:28.
  5. Before the LLM lane starts, one RunBudget is constructed from the per-run dollar ceiling (packages/atif-analytics/src/atif_analytics/application/analyze.py:140); all five LLM stages then go through one call site, each receiving the shared reader and that budget, and a stage that finds the budget exhausted is skipped with nothing stamped — :181.
  6. A stage drops the sessions whose checkpoint row still matches their mtime and last step timestamp, so a re-run costs nothing for unchanged work — packages/atif-analytics/src/atif_analytics/infrastructure/sqlite_state/checkpointer.py:136.
  7. The concrete provider satisfies LlmStructuredProvider structurally, and its synchronous inner method is the only place a Bedrock invoke_model call is issued, under tenacity retry with token usage accumulated against the budget — packages/atif-models/src/atif_models/infrastructure/openai_bedrock.py:258.
  8. Results land as sharded part-<ns>.parquet files under <corpus_root>/analytics/ (packages/atif-analytics/src/atif_analytics/infrastructure/parquet_cache.py:90), and each completed session is upserted into the SQLite WAL checkpoint — packages/atif-analytics/src/atif_analytics/infrastructure/sqlite_state/checkpointer.py:179.
loop [each LLM stage under one RunBudget] run_analyze(settings, dry_run=false) CorpusReader.load_steps (json over trajectory.json) StepEvent rows load_embeddings for cluster / community uuids + vectors write cluster / terms / community parquet filter_unchanged against state.db classify_structured(prompt, schema) invoke_model (tenacity retry) structured JSON + usage parsed rows, cost accrued write_part parquet, mark_completed per-stage summary dict atif-cli atif-analytics corpus disk Lance store atif-models Bedrock

See also

Section titled “See also”