Uploading files

You post a file and, with nothing else to run, it becomes an answer to a question. That path is two workflow steps and one function call to install them, and a flag on the same call adds a second index that makes the upload connected as well as findable.

Installing the pipeline

POST /files hashes the bytes, stores them content-addressed in object storage, registers a resource, and starts the ingestion workflow named by the channel you uploaded to. That workflow has to exist first.

Here we install the pipeline that turns any arriving file into searchable chunks.

The model has to be registered before the pipeline can name it, so this is two statements and the first one is not optional. This one is bge-small-en-v1.5, which the ingest worker runs in-process through fastembed: the endpoint is the model id the worker loads, and there is no key to name. A hosted model is the same insert with provider openai, its url as the endpoint and credential_ref => 'LLM_API_KEY'. on conflict because the harbour sample registers this same model — if you followed the install guide it is already there, and the insert is then a no-op rather than a duplicate-key error on the first line of this page:

insert into aiq.embedding_models (name, dim, provider, endpoint, credential_ref, is_default)
values ('bge-small-en-v1.5', 384, 'fastembed', 'BAAI/bge-small-en-v1.5', null, true)
on conflict (name) do nothing;
select aiq.register_embedding_space('bge-small-en-v1.5');

select content.install_ingest_workflow('bge-small-en-v1.5');
-- ingest_file: parse, embed with bge-small-en-v1.5 in batches

Skip the first two and the third refuses rather than installing something that would fail later, which is the behaviour you want but does read as an error on a fresh install:

ERROR:  embedding model 'bge-small-en-v1.5' is not registered, so an
        ingestion pipeline naming it would fail at every run. Register it
        first, or call this with no argument to install the parse-only pipeline.

Both of the steps it writes run on the ingest queue, so something has to be polling that queue. The compose file ships an ingest-worker service for exactly this; a deployment that runs only the http worker accepts uploads and never reads them — POST /files returns 201, the run starts, and parse sits at ready with nothing erroring.

Why it works — it writes an ordinary workflow document, and no step carries a url

name: ingest_file
steps:
  - id: parse
    work: true
    queue: ingest
    input: {op: ingest, resource_id: '{{run.resource_id}}'}
  - id: embed
    needs: [parse]
    work: true
    queue: ingest
    input: {op: embed, resource_id: '{{run.resource_id}}',
            model: bge-small-en-v1.5}

You can read it, edit it or replace it — it is a definition like any other, which is the point of generating one rather than hard-coding a pipeline.

Neither step carries an endpoint. The worker asks aiq.embed_batch_call() for the url, the request shape, the path to the vector in the response and the name of the credential, so the model lives in exactly one place and naming it here is naming a model that is registered.

Uploading one

Here we put a file in and get chunks out, with nothing to run afterwards.

POST /files takes the bytes as the body — the Content-Type is what picks the family — and the Content Server is on port 8081 in the compose stack.

curl -s http://localhost:8081/files \
  -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: text/markdown' \
  -H 'x-p8-title: Aurora Kestrel note' \
  --data-binary @note.md

201, then the run the upload started

{"resource_id":"44057672-…","checksum":"b440dc19…","bytes":144}

 step_key | queue  |  status   | attempts
----------+--------+-----------+----------
 parse    | ingest | succeeded |        1
 embed    | ingest | ready     |        4

 resources | chunks
-----------+--------
         1 |      1

x-p8-org decides who besides you can read the bytes back. Without it, a resource is stored on the shared tier with visibility = 'private', so it is yours alone: another user's query of content.resources does not return it, and GET /files/{resource_id} resolves it only for you. With it, the resource belongs to that org and every member of it may read it:

  no x-p8-org       stored on the shared tier, visibility private
                    anonymous GET -> 404 "no such file, or not yours"
                    as another user -> 404, the same answer
                    as you -> 200, the bytes
  x-p8-org: <a>     stored with org <a>, visibility org
                    anonymous GET -> 404 "no such file, or not yours"
                    as tenant <b> -> 404, the same answer
                    as tenant <a> -> 200, the bytes

The wrong tenant and the missing file are indistinguishable, by design: a 404 that meant "exists, not yours" would confirm the resource id.

embed retrying is what a hosted model with no key looks like: parsing and chunking need no model, so the file is chunked and text-searchable, and the vector arrives once LLM_API_KEY is set on the ingest worker. The bundled bge-small-en-v1.5 needs no key, so on the default install embed succeeds. x-p8-channel picks the channel whose workflow runs — omitted, it is the default one — and $TOKEN is the one from install.

What we do with each format

The family comes from the content type and falls back to the extension. We never take application/octet-stream as a family, since that is what an uploader sends when it does not know.

Format	What reads it
markdown, text	itself, already in the intermediate form
html	the standard library's `HTMLParser`
pdf	`pymupdf4llm`, which reads the PDF's own structure, no model, milliseconds a page
docx	`python-docx`, a zip and some XML in document order
wav, mp3, m4a	a transcription endpoint, since a model call belongs to the worker
csv, tsv, xlsx, parquet	`pyarrow`, and the file becomes a Parquet dataset rather than chunks
images, anything unrecognised	stored and retrievable, not pretended to be text

Why it works — everything text-shaped normalises to markdown before it is chunked

Chunking each format from its own structure would be three chunkers with three sets of edge cases. Normalising first means one chunker, and a passage that still carries the heading and the table that told a reader what the sentence was about.

We chunk with semchunk, which splits on the most significant boundary that fits the budget — sections, then paragraphs, then sentences, then clauses — rather than on a separator list. On Isaacus's legal retrieval measurement that is worth about 8% over recursive splitting and about 12% over fixed-size chunks, it is pure Python with no model behind it, and it returns offsets, so a citation can point at a span rather than at a whole chunk.

The default is 700 tokens, and most RAG code uses 200 to 400. That is the wrong size for what we do with a chunk, which is retrieve it, show it to a model and cite it: a 250-token chunk retrieves the sentence that matched and loses the sentence that explains it, so the model answers from a fragment and the citation points at one. 700 tokens is two or three paragraphs of prose. Transcripts default to 450 with more overlap, because there are no headings and no blank lines to cut on.

The token counter is characters ÷ 4 and works offline. tiktoken is a better count and downloads its encoding the first time it runs, which is an unhappy thing to discover inside a container with no egress, so it is opt-in: tokenizer: "tiktoken:cl100k_base".

A table is not prose

Here we make a spreadsheet answerable by query rather than by retrieval.

select content.dataset_uri('incidents');
-- s3://p8-content/datasets/incidents/*.parquet

tabular.read_dataset(uri, "select site, sum(downtime_hours) from dataset group by 1")

Why it works — chunking a CSV is the default elsewhere and it answers nothing

Two hundred rows of a price table become two hundred near-identical embeddings that rank against each other, so what was August revenue retrieves five rows that all look equally like the question. A table already has a query language.

So a tabular upload converts once to Parquet, lands beside the original bytes in object storage, and is registered as a dataset with a chunk count of zero. The column list comes from the Arrow schema the conversion produced rather than from a CSV header, since a header is names and a query needs types.

Files for one dataset land under one prefix and mode: append means the reader globs it, so this month's export joins last month's under one name. That is the whole of the resemblance to Iceberg: there is no manifest, no snapshot isolation, no schema evolution and no time travel. What breaks first is a schema change between two files under one prefix, and a deployment that needs those things needs Iceberg — which it can reach with a read_parquet rather than a rewrite.

Changing the policy without changing the code

Here we make every file on one channel chunk small, without touching the worker.

update content.channels
   set config = content.merge_policy(config, '{"ingest_policy":
        {"chunking": {"target_tokens": 300}, "keep_markdown": false}}'::jsonb)
 where name = 'contracts';

Why it works — three layers, and a later one wins key by key at every depth

family default   a fact about a format and its library, in code
      ↓
channel          content.channels.config.ingest_policy
      ↓
upload           content.resources.metadata.ingest_policy

Defaults are statements about libraries and change when the library does, in the same commit. Overrides are statements about a deployment's content, made by the person uploading — who is usually not the person who deployed the worker.

The merge is deep, because a shallow one lets an upload that sets a single chunking field erase the channel's other chunking fields, and the uploader would then be running a strategy nobody chose.

The policy is validated strictly, so a misspelled key is refused by name rather than ignored, and it fails the step terminally — it will not validate on the fifth attempt either.

Embedding a corpus is one call, not one per chunk

Here we embed four hundred chunks without four hundred task rows.

{"embedded": 412, "requests": 5}

Why it works — the payload cap makes the per-chunk shape impossible, not merely slow

Embedding a query is a step that produces a vector and hands it to the next step, and that is what the embed: step kind is for. Embedding a corpus runs the other way: N vectors that have to be kept, keyed by chunk, in the space of the model that produced them.

It cannot go through the engine one vector at a time. A step's output is capped at 64KB and one 1536-dimension vector from a hosted model is about 31KB of JSON, so two vectors do not fit in one task output whatever else you do. The process holding the vectors is therefore the one that writes them, and what comes back to the engine is a receipt.

For an eight-chunk document that is one request and one task row, where the per-chunk shape was eight of each. Re-embedding a chunk replaces its vector, so re-ingesting a document whose text changed does not leave the old vector ranking against the new text.

A second index off the same parse

Chunks and vectors make an upload findable. The other thing we can do with the same parse is make it connected: read what each passage names and how those things relate, and land that as nodes and edges in the graph.

Here we turn the graph index on, which is a flag on the same install call.

select aiq.install_structure_null('gpt-4o-mini');
select content.install_ingest_workflow(
           p_model       => 'bge-small-en-v1.5',
           p_graph_index => true,
           p_graph_model => 'gpt-4o-mini');
-- ingest_file: parse, embed with bge-small-en-v1.5 in batches,
--             graph-index each window with structure_null (gpt-4o-mini)

every extracted edge carries where it came from

    relation     |       title        |     agent
-----------------+--------------------+----------------
 contains        | r7-field-notice.md | structure_null
 affiliated_with | r7-field-notice.md | structure_null
 supplies        | r7-field-notice.md | structure_null

Why it works — the two indexes are siblings, and the extractor is a row

The flag adds two steps:

  - id: graph
    needs: [parse]                 # a sibling of embed, not a successor
    matrix:
      rows: {function: windows_to_index, args: ['{{run.resource_id}}']}
      max_fanout: 200
      template: {queue: http, rate_key: structure_extraction, rest: …}

  - id: land_graph
    needs: [graph]
    sql: {function: land_graph_windows, args: ['{{steps.graph.result.task_id}}']}

embed and graph both wait on parse and on nothing else, so the two indexes run at the same time over the same text, a deployment with no embedding model can still build a graph, and an extraction that fails costs you no vectors. A window here is a chunk, so the extractor reads the same passages the embedder embedded and a node can be traced back to the passage that named it.

The extractor is a row in agentic.agents rather than a prompt in the workflow, which means changing what it looks for is an update rather than an edit to every pipeline that uses it. aiq.install_structure_null() writes the default one: a system instructions asking for what the document names and how those things connect, and a JSON Schema whose relation field is an enum built from aiq.graph_vocabulary. The vocabulary is closed, so an extractor cannot invent a relation the graph then has to carry forever.

The five uploaded files produced about 90 soft nodes and 70 extracted edges over 13 relations. Those counts move between runs, since a model is doing the reading. What does not move is the provenance above.

Why it works — two things to know before you turn it on

It costs a completion per window, where the embedding branch costs a fraction of a cent per document, and that is why the flag defaults to off. On five documents it was a few cents; on a corpus it is the line item to watch.

A cheap model gets some of it wrong, and we drop those rather than fail the upload. Two kinds: an edge pointing at node 8 when the model listed eight nodes, and a relation outside the vocabulary — gpt-4o-mini emitted trips on a document about a robotic arm.

The schema carries the vocabulary as an enum, and OpenAI honours an enum only in strict structured-output mode, which this schema cannot use because strict wants every property in required and the attribute bag is optional. So the enum is a strong hint rather than a constraint. Both kinds are dropped before landing and counted in the receipt (edges_dropped_ordinal, edges_dropped_offvocab), which also tells you when the vocabulary is too small for your corpus.

Extraction lands names, so Ravensworth and Ravensworth Precision are two nodes until something decides they are one. aiq.dream(), below, promotes a soft node only when its name is exactly a registered node's key; Ravensworth Precision is not, so it stays a judge's call. The graph answers what did the documents name, and how did they connect it rather than what is out there.

Memory from a conversation

A file is one way facts arrive. A conversation is the other: what a person tells an agent is worth keeping, and the same extractor can read it.

Here we install the workflow that reads a finished conversation into the graph, and look at what it leaves behind.

select agentic.install_reflect_workflow('gpt-4o-mini');
-- reflect_session: each message the user wrote, graph-indexed with structure_null (gpt-4o-mini)

Start it for a session once the conversation is over. It runs as an ordinary workflow, so the run is a trace like any upload's:

select workflow.start_workflow('reflect_session', '{"session_id": "<the session>"}');

Every fact it writes names the session and the message that said it, and holds from when it was said rather than from when it was read. That is enough to put the memory, the fact and the words side by side:

select m.created_at as said_at, m.content as said, e.relation,
       src.summary as subject, dst.summary as object, e.valid_from
  from aiq.edges e
  cross join lateral jsonb_array_elements(e.provenance->'sources') s
  join agentic.messages m on s->'message_ids' ? m.id::text
  join aiq.nodes src on src.id = e.src_id
  join aiq.nodes dst on dst.id = e.dst_id
 order by m.created_at;

Why it works — a window per thing the person said, scoped like the session

reflect_session is the upload's graph branch pointed at a conversation: a matrix over agentic.session_windows(session_id), the same extractor row, the same land_graph_windows. A window is one message the person wrote; the agent's replies are context and are not read, so an agent saying "you live in Leeds" never becomes the fact.

The session scopes what lands, the way a document's resource does: a private conversation's facts are its owner's, a conversation shared with an org is readable by the org, and one made private again takes its facts with it, because every read of an edge asks whether the caller may still read the session it came from. Reflecting the same conversation twice adds no second source. (REM-969.)

Reconcile what knit landed

Knit writes every session's facts as it hears them, so the graph holds a person's old address beside the new one. Dream reconciles the part of that which is known by construction, and leaves the rest for a judge.

Here we declare that a person lives in one place at a time, write two sessions, and let dream end the older fact.

select aiq.register_graph_vocabulary('[{"relation":"lives_in","single_valued":true}]');
select aiq.upsert_graph('{"nodes":[{"k":"person","n":"Maya"},{"k":"place","n":"Leeds"}],
  "edges":[{"s":0,"t":1,"r":"lives_in"}]}', '{"agent":"knit","run_id":"s1"}');
select aiq.upsert_graph('{"nodes":[{"k":"person","n":"Maya"},{"k":"place","n":"Manchester"}],
  "edges":[{"s":0,"t":1,"r":"lives_in"}]}', '{"agent":"knit","run_id":"s2"}');
select aiq.dream() - 'since' - 'until' as receipt;

the receipt, then Maya's lives_in edges

                                          receipt
--------------------------------------------------------------------------------------------
 {"superseded": 1, "promoted_exact": 0, "promotion_refused": 0, "soft_left_for_a_judge": 3}

     lives_in     | current
------------------+---------
 place/leeds      | f
 place/manchester | t

To run it on the engine's clock instead of by hand, select aiq.schedule_dream('*/15 * * * *'); puts a dream workflow on workflow.schedules. Nothing is scheduled until you do.

Why it works — dream does only what it could show its working for

Supersession needs a declaration. single_valued is a property of the relation, set through register_graph_vocabulary, and null until you say. For a relation declared true, every current edge but the newest, per subject and org, ends where the next one begins. The older edge keeps its provenance, so an as-of read before the move still finds Leeds. prefers is not single-valued: a person prefers many things, and which preference replaced which is a judgement. On a 48-fact benchmark, declaring the four relations that hold one value recovered 7 of its 11 supersessions with none invented; treating every relation as single-valued recovered 8 and invented 22.

Promotion needs an exact key. A soft node is promoted when its name is exactly the canonical key of one registered node of the kind it claimed, in its org, and the resolution is recorded at tier registry. An alias such as "P. Raman" is not exact, so dream counts it in soft_left_for_a_judge rather than guessing.

A second run does nothing. Each run is a row in aiq.dream_runs, and the next starts where the last ended.

Where it fails loudly

Three of these exist because the alternative is a pipeline that reports success.

A parse that yields nothing raises, and names the likely cause for the parser that ran. A scanned PDF returning "" would otherwise produce zero chunks, a resource that looks stored, and a file absent from every search anyone runs.
ready means "in the corpus". A resource is marked ready only once chunks were written and stored otherwise, which is what content.check_drift() reconciles against.
A missing vector raises rather than landing the rest of the batch, since a chunk with no vector is a hole in a corpus that still answers questions.

What is not built, and one thing that bites

Uploading a file whose title matches one already there fails to ingest. A resource projects into the node registry under coalesce(title, uri, id) and those keys are unique, so v2 of report.pdf raises inside the projection and the parse step fails for a reason that has nothing to do with parsing. This is the most ordinary thing a user does. The fix is a decision about identity rather than about ingestion — either the projection tolerates a taken name, or document keys stop being titles — and until it is made, give the second upload a different title.

The rest of the gaps:

No OCR. A scanned PDF fails loudly and stays pending. Docling or a vision model per page is the opt-in, and it is not written.
No image understanding. Images are stored. Describing one with a vision model is a real feature, and defaulting to it would put a model call behind every avatar upload.
Re-chunking is not incremental. Fixing a typo in a 100-page document re-chunks and re-embeds all of it.
A failed batch re-embeds the whole batch. The step is idempotent and not resumable, so four of five requests already paid for are paid for again.
A provider that cannot batch is called N times. Ollama's embeddings endpoint takes one prompt, so the worker loops — same receipt, same retry, and no faster than the per-chunk shape it replaced.

Five formats — markdown, PDF, DOCX, WAV and CSV — were uploaded through a live stack with both indexes on, and four questions were answered afterwards from four different formats, with the CSV answered by SQL over Parquet rather than by retrieval. Everything above is that run: the chunk counts, the one request for eight chunks, the edges above and the relation the model invented.

Next: querying.