Corpus

The public façade. One class, ~40 methods, the only thing most Python users need to import.

from kglite_docs import Corpus

with Corpus.open("kb.kgl") as c:
    hits = c.search("query")

The PDF knowledge base. Light wrapper over Store + an embedder.

close

close() -> None

Persist + drop in-process state. Mostly useful at the tail of a with block (which calls this automatically) or to release the embedder's ONNX session early on long-lived processes.

ingest

ingest(
    path: str | Path | None = None,
    *,
    text: str | None = None,
    title: str | None = None,
    source_uri: str | None = None,
    metadata: dict[str, object] | None = None,
    format: str | None = None,
    embed: bool = False,
    structure_aware: bool = False,
    context_summary: str = "",
    source_party: str = ""
) -> IngestResult

Ingest a document. Three modes:

• ingest("paper.pdf") — file path; format auto-detected from the extension. Pass format= to override.
• ingest(text="# Notes\n…", title="my-notes") — raw text / markdown. Useful for agent-generated synthesis articles.
• ingest("doc.bin", format="md") — file path with explicit format hint when the extension doesn't match.

Embedding is opt-in. By default (embed=False) ingest does not touch the embedding model — it parses, chunks, and writes the graph, leaving ready chunks :Unembedded. Call :meth:index afterwards (or pass embed=True here) to compute vectors and enable :meth:search. Non-semantic workflows (browse, cypher, tag, review, OCR, export, translate) need no embeddings at all.

With structure_aware=True chunking starts a fresh chunk at every top-level heading (never packing or overlapping across one) — cleaner Section boundaries and pinpoint cites; the default packs greedily.

context_summary (opt-in) is a document-level blurb prepended to each chunk before embedding so the vector carries global context (mitigates cross-document speaker/source confusion); the stored chunk text is unchanged. You supply the summary (e.g. from an LLM pass) — none is generated here.

Returns an :class:IngestResult with the assigned doc_id (sha256 of file or text bytes), chunk count, OCR-pending page count, and how many chunks were embedded (0 unless embed=True).

Raises :class:UnsupportedFormatError for unknown formats and :class:IngestError for parse failures.

ingest_dir

ingest_dir(
    directory: str | Path,
    *,
    recursive: bool = True,
    patterns: list[str] | None = None,
    embed: bool = False,
    structure_aware: bool = False
) -> list[IngestResult]

Ingest every supported file under directory. By default scans for all known formats: PDF, DOCX, PPTX, MD, HTML, TXT, and common image formats.

As with :meth:ingest, embedding is opt-in (embed=False). For bulk loads prefer the default and call :meth:index once at the end — one batched embedding pass beats per-file embedding.

count_unembedded

count_unembedded(*, doc_id: str | None = None) -> int

How many ready chunks are still awaiting embedding. Scope with doc_id. Tracked by the c.embedded boolean property (not a removable label — see :meth:index).

index

index(
    *,
    doc_id: str | None = None,
    batch_size: int = 16,
    max_chunks: int | None = None,
    max_seconds: float | None = 30.0
) -> dict[str, Any]

Embed ready-but-unembedded chunks — the optional second phase of ingestion that makes :meth:search work.

Bounded and loop-friendly. A single call does at most max_seconds of work (wall-clock budget, default 30s) or max_chunks chunks, whichever is hit first, then commits what it embedded and returns pending > 0 if more remain. This keeps any one call comfortably under an MCP client's per-call timeout even for a large multi-document corpus — the caller loops until pending == 0::

while corpus.index()["pending"]:
    ...

Pass max_seconds=None (and max_chunks=None) to drain everything in one call when there's no timeout to worry about (e.g. CLI preload). Idempotent: only touches chunks not yet embedded (tracked by the c.embedded property), so looping or re-running is safe. Scope to one document with doc_id.

Chunks are embedded in length-sorted batches so each batch pads to a similar sequence length (bge-m3 caps at 8192) — avoids a few long chunks inflating the padding for a document's worth of short ones. The :Embedded label is added (never removed) as chunks are indexed; the pending side is the property, because kglite's remove_label leaves the label-predicate index stale.

Returns {"embedded": n, "pending": remaining, "doc_id": ...}.

set_source_party

set_source_party(doc_id: str, party: str) -> dict[str, Any]

Tag a document with its source party (who produced/filed it) and inherit it to the document's chunks — so an admission against interest (primary text by the adverse party) can be surfaced. party is free-text (available_source_parties() lists the registered set).

available_source_parties

available_source_parties() -> list[dict[str, str]]

Registered source-party values (value + label + description). Empty until a schema pack registers them; any value is still accepted.

list_sections

list_sections(doc_id: str) -> list[SectionRow]

Sections of a document (the grain between document and chunk), in reading order, each with its chunk_count. Sections are derived at ingest from the PDF outline or top-level headings; re-ingest documents ingested before this feature to populate them.

compare_documents

compare_documents(
    doc_a: str,
    doc_b: str,
    *,
    queries: list[str],
    top_k_per_query: int = 5,
    max_tokens_per_query: int = 2000,
    agent_id: str | None = None
) -> ComparisonResult

Side-by-side cross-document retrieval.

For each query, returns the top hits from doc_a and doc_b independently, plus a budgeted merged context bundle ready to hand to a downstream LLM that's writing a comparison.

queries is a list of axes you want to compare on — e.g. ["retrieval architecture", "training objective", "evaluation metrics"] for two papers. 3–7 queries is the sweet spot.

Returns a ComparisonResult dict:

{
  "doc_a_id": ..., "doc_a_title": ...,
  "doc_b_id": ..., "doc_b_title": ...,
  "queries": [
    {
      "query": "...",
      "doc_a_hits": [...],   # top_k_per_query
      "doc_b_hits": [...],
      "merged_context": {used_tokens, items}   # ComposedContext
    },
    ...
  ]
}

register_agent

register_agent(
    agent_id: str,
    *,
    kind: AgentKind = "llm",
    model: str = ""
) -> dict[str, Any]

Idempotent lazy registration. Bumps last_seen + counters if the agent already exists; minimal-record creates otherwise. Does not overwrite template fields — use upsert_agent for that.

upsert_agent

upsert_agent(
    agent_id: str,
    *,
    kind: AgentKind = "llm",
    model: str = "",
    role: str = "",
    system_prompt: str = "",
    tools: list[str] | None = None,
    context: dict[str, Any] | None = None,
    description: str = ""
) -> AgentConfig

Write the agent's template: role, system prompt, model, tool list, free-form context. Field-level merge with whatever already exists. Returns the resulting config.

Once defined, fetch with get_agent(agent_id) and use the config to launch your LLM call — the agent_id you then use for subsequent add_summary / complete_review / etc. will be attributed back to this template.

get_agent

get_agent(agent_id: str) -> AgentConfig

Full agent config — template + counters. Empty dict if the agent isn't registered yet.

list_agents

list_agents(
    *,
    role: str | None = None,
    kind: AgentKind | None = None
) -> list[AgentRow]

List configured agents, optionally filtered by role or kind.

agent_activity

agent_activity(
    agent_id: str,
    *,
    target_id: str | None = None,
    limit: int = 50
) -> AgentActivity

Everything this agent has done in the corpus — optionally scoped to one target node. Buckets: views, summaries, tags, translations, review_events, verification_events.

status

status() -> CorpusStatus

One-call snapshot: docs, pages, chunks, embedded/unembedded, image_pages, pending_ocr, studies. The first thing to check.

coverage_report

coverage_report(
    *, doc_id: str | None = None
) -> CoverageReport

Honest extraction + embedding coverage per document + corpus-wide, with a human-readable summary — what's image-only / low-text (unanalyzed unless OCR'd) and how many chunks are unembedded (search blind until index()). Pass doc_id to scope the per-doc rows.

triage_map

triage_map(*, doc_id: str | None = None) -> TriageMap

One cheap call that aggregates the deterministic content signals — the content_kind breakdown, boilerplate / low-quality counts, structured- entity coverage, element-classification coverage, embedding state, OCR-pending pages — so an agent orients without reading the corpus. Scope with doc_id.

element_coverage

element_coverage(
    element: str,
    *,
    doc_id: str | None = None,
    section_id: str | None = None
) -> dict[str, Any]

How an element= scope partitions the ready chunks (in_scope, excluded_other_element, excluded_unclassified, ready_total), with in_scope + excluded_total == ready_total. The honest-coverage block a scoped study_ledger also embeds; unknown element raises.

element_consistency

element_consistency() -> dict[str, Any]

Audit element labels vs the canonical element_types_json ({checked, inconsistent, sample}) — surfaces any label/property drift.

next_unclassified

next_unclassified(
    *,
    doc_id: str | None = None,
    section_id: str | None = None,
    agent_id: str | None = None,
    limit: int = 20,
    ttl_seconds: int = 1800
) -> list[dict[str, Any]]

Ready chunks not yet classified, in reading order. With agent_id atomically claims them (punchcard, disjoint from study claims); without it, a read-only preview. Classify once, then route many studies by element=.

classify_chunk

classify_chunk(
    chunk_id: str,
    *,
    elements: list[str],
    agent_id: str,
    model: str = "",
    confidence: float | None = None
) -> dict[str, Any]

Classify a chunk into zero or more registered element types (load a schema pack first, e.g. schemas.load_schema('legal')). Empty elements = a deliberate "no element applies" → :Unclassified. Add-only labels (recall-safe); a divergent second agent adds :Contested.

classify_many

classify_many(
    items: list[dict[str, Any]],
) -> dict[str, Any]

Batch-classify many chunks. Each item: {chunk_id, elements, agent_id} (+ optional model, confidence).

ocr_status

ocr_status(*, doc_id: str | None = None) -> OcrStatus

Coverage summary: which documents have un-OCR'd pages, and what fraction of the corpus is still pending. Pass doc_id to narrow to one document.

submit_ocr_many

submit_ocr_many(
    rows: list[dict[str, Any]],
    *,
    agent_id: str = "ocr",
    model: str = ""
) -> dict[str, Any]

Submit many pages' OCR at once (each row {page_id, markdown} or {page_id, tiles}). rows is a structured argument the SDK escapes — so agents don't hand-serialize multi-line verbatim text into a fragile JSON file. A failing row is reported, not fatal.

export_ocr

export_ocr(
    doc_id: str, *, out_path: str | None = None
) -> dict[str, Any]

Write a document's OCR to a sidecar JSON (<source>.ocr.json) — portable, auditable, hand-correctable, re-importable. Carries each page's ocr_status/legible_chars.

import_ocr

import_ocr(path: str) -> dict[str, Any]

Round-trip a sidecar JSON back in (apply each page via submit_ocr). The document must already be ingested (matched by doc_id).

list_illegible_pages

list_illegible_pages(
    *,
    doc_id: str | None = None,
    limit: int = 50,
    include_images: bool = False,
    dpi: int = 200
) -> list[dict[str, Any]]

Pages OCR'd but illegible/partial (effectively unreadable) — the worklist for human review or a stronger-model retry via request_ocr(force=True). Optional include_images renders each.

request_ocr

request_ocr(
    *,
    page_id: str | None = None,
    doc_id: str | None = None,
    page_number: int | None = None,
    agent_id: str,
    agent_type: str = "",
    dpi: int = 200,
    force: bool = False
) -> dict[str, Any]

Lazy OCR: hand back the OCR task (rendered page + verbatim prompt) for a needs_ocr page so the agent transcribes it and calls submit_ocr. Identify by page_id or doc_id+page_number; agent_type is echoed so an orchestrator can route to a specific OCR subagent. force=True re-OCRs an already-transcribed page (escalate an illegible result to a stronger model) — the new submit replaces the page's chunks.

submit_ocr

submit_ocr(
    page_id: str,
    markdown: str = "",
    *,
    agent_id: str,
    model: str = "",
    confidence: float | None = None,
    tiles: list[dict[str, Any]] | None = None
) -> dict[str, Any]

Patch an agent's transcription back into a page. Pass whole-page markdown, or tiles=[{tile_index, markdown}] from a tiled request_ocr (stitched in order). Records the legibility ocr_outcome.

verify_claim

verify_claim(
    claim_text: str,
    *,
    against_chunk_ids: list[str] | None = None,
    top_k: int = 5
) -> dict[str, Any]

Find chunks that support a free-text claim via vector search.

Deprecated: prefer the study flow (define_study → assess → study_ledger) to evaluate a claim across chunks — it's richer (for/against + weight + provenance), multi-agent, and verifiable. This one-shot helper remains for quick checks.

add_translation

add_translation(
    chunk_id: str,
    target_lang: str,
    text: str,
    *,
    agent_id: str,
    model: str = "",
    status: TranslationStatus = "draft"
) -> str

Store an agent-produced translation for a single chunk.

assemble_translated_document

assemble_translated_document(
    doc_id: str,
    *,
    target_lang: str,
    prefer_reviewed: bool = True
) -> dict[str, Any]

Stitch a document's translated chunks back together. Pages without a translation fall back to the original text.

enqueue_review

enqueue_review(
    target_id: str,
    *,
    target_kind: TargetKind = "Chunk",
    priority: int = 0,
    note: str = "",
    enqueued_by: str = "system"
) -> str

Add a target node (chunk/summary/document/page) to the review queue. Returns the ticket id.

enqueue_chunks_for_review

enqueue_chunks_for_review(
    *,
    doc_id: str | None = None,
    status_filter: str | None = "ready",
    priority: int = 0,
    enqueued_by: str = "system"
) -> dict[str, Any]

Bulk-enqueue every chunk (optionally scoped to one document or a Chunk.status filter). Skips chunks that already have a ticket.

claim_review

claim_review(
    ticket_id: str, *, agent_id: str
) -> dict[str, Any]

Atomically claim a specific ticket. Raises ReviewConflict if it's not currently in the new state.

claim_next_review

claim_next_review(
    *,
    agent_id: str,
    target_kind: TargetKind | None = None,
    min_priority: int | None = None
) -> ReviewTicketDetail | None

Atomic 'pull from the queue': finds the highest-priority new ticket and claims it for agent_id. Returns the ticket with the target hydrated, or None if the queue is empty.

unclaim_review

unclaim_review(
    ticket_id: str, *, agent_id: str, reason: str = ""
) -> dict[str, Any]

Release a claim without a verdict. Only the current claimer can unclaim.

complete_review

complete_review(
    ticket_id: str,
    *,
    agent_id: str,
    verdict: ReviewVerdict = "reviewed",
    accuracy: float | None = None,
    authenticity: str | None = None,
    notes: str = "",
    tags: list[str] | None = None
) -> dict[str, Any]

Mark a ticket reviewed. verdict is one of reviewed / needs_revision / rejected. Optional accuracy (0-1) and authenticity capture the agent's judgement. tags are applied to the target chunk (only when target_kind=Chunk).

list_review_queue

list_review_queue(
    *,
    status: ReviewStatus | None = None,
    target_kind: TargetKind | None = None,
    agent_id: str | None = None,
    limit: int = 50
) -> list[ReviewTicketRow]

List tickets with their current event-sourced status.

get_review_ticket

get_review_ticket(
    ticket_id: str,
    *,
    with_target: bool = True,
    with_events: bool = True
) -> ReviewTicketDetail | None

Full ticket detail including the target node and the immutable event audit trail.

review_stats

review_stats() -> ReviewStats

Kanban board summary: counts per status + per-agent in-review.

define_study

define_study(
    question: str,
    *,
    created_by: str,
    title: str | None = None,
    status: StudyStatus = "open"
) -> str

Create a Study (a question/claim to gather evidence for/against). Returns the study id. See assess / study_ledger / verify_assessment.

assess

assess(
    study_id: str,
    chunk_id: str,
    *,
    stance: Stance,
    weight: float,
    agent_id: str,
    rationale: str = "",
    model: str = "",
    provenance: Provenance = "primary_text",
    quote: str = "",
    char_start: int | None = None,
    char_end: int | None = None,
    context_chunk_ids: list[str] | None = None
) -> dict[str, Any]

Record stance (supports/against/neutral/deferred) + probative weight [0,1] + rationale on a chunk toward a study. Append-only; never embeds. deferred = read but unjudgeable yet (blocked/needs evidence): counted distinctly and kept in the work-list for a later pass.

provenance records what was checked (the basis, vs weight the strength): primary_text (read the source — default), characterization (a paraphrase/summary), or scanned_unread (an unread scan; provisional). Surfaced per row in study_ledger.

quote/char_start/char_end are an optional pinpoint span — the exact passage the call rests on, surfaced in the ledger for pinpoint cites. Validated against the chunk text (out-of-range / quote-not-found rejected).

context_chunk_ids: neighbor chunks read to interpret the focal one; recorded so retrieval pulls the span and they're excluded from the work-list (no double-judging).

assess_many

assess_many(
    study_id: str, rows: list[dict[str, Any]]
) -> dict[str, Any]

Batch-assess many chunks in one validated, batched write (a single persist through the MCP layer). Each row is a dict with chunk_id/stance/weight/agent_id (+ the optional assess fields). One bad row aborts the whole batch — nothing is written.

supersede_assessment

supersede_assessment(
    old_id: str,
    *,
    stance: Stance,
    weight: float,
    agent_id: str,
    rationale: str = "",
    model: str = "",
    provenance: Provenance = "primary_text",
    context_chunk_ids: list[str] | None = None
) -> dict[str, Any]

Audit-preserving correction: record a new assessment that explicitly supersedes old_id (a SUPERSEDES edge). The old one is kept but hidden from study_ledger by default — resolving cross-agent corrections to a single current row per chunk. Inherits the old assessment's study+chunk.

study_ledger

study_ledger(
    study_id: str,
    *,
    stance: Stance | None = None,
    min_weight: float | None = None,
    verified_only: bool = False,
    doc_id: str | None = None,
    section_id: str | None = None,
    element: str | None = None,
    include_superseded: bool = False,
    limit: int = 200
) -> Ledger

Weight-ranked evidence ledger for a study + support/against tallies. Pass stance="supports"/"against" to retrieve just that side, or doc_id=/section_id= to scope to one document or section. Current-by-default: superseded assessments are hidden unless include_superseded=True (each row carries a superseded flag). The result reports total (matches before limit) and returned; total > returned means it was clipped.

verify_assessment

verify_assessment(
    assessment_id: str,
    *,
    verdict: AssessmentVerdict,
    verifier_agent_id: str,
    notes: str = "",
    provenance: Provenance | None = None
) -> dict[str, Any]

Second-agent check of an assessment: verified / disputed / duplicate. Self-verification is rejected. provenance (optional) records what the verifier checked — stored on the verification event.

synthesize_study

synthesize_study(
    study_id: str, *, agent_id: str, note: str = ""
) -> dict[str, Any]

Mark the cross-chunk synthesis pass as run (clears the conclude gate). The agent reads the whole ledger + records cross-chunk Findings first; see synthesis_prompt() for what to hunt.

synthesis_prompt

synthesis_prompt() -> str

The prompt an agent reads before the synthesis pass — the domain-neutral hunt list plus any registered domain addenda.

conclude_study

conclude_study(
    study_id: str,
    text: str,
    *,
    agent_id: str,
    model: str = "",
    embed: bool = False,
    acknowledge_no_synthesis: bool = False
) -> str

Write a conclusion (stored as a verifiable Summary on the Study). Refuses (SynthesisRequiredError) unless the study has been synthesized, unless acknowledge_no_synthesis=True records an audited skip.

list_studies

list_studies(
    *,
    status: StudyStatus | None = None,
    created_by: str | None = None
) -> list[StudyRow]

List studies that have been run (newest first).

get_study

get_study(study_id: str) -> StudyRow | None

Study metadata + tallies + its conclusion summaries.

study_conflicts

study_conflicts(study_id: str) -> ConflictReport

Chunks with both a current supports and against assessment — the contested evidence to review first. Computed over the current (non-superseded, latest-per-agent) set; each conflict carries its opposing rows split by side.

study_semantic_conflicts

study_semantic_conflicts(study_id: str) -> dict[str, Any]

Cross-chunk contradictions: within a classified element/topic, different chunks carrying opposing stances (the disparate-treatment / conflicting-disposition class same-chunk study_conflicts can't see). Reports honest coverage (checked vs skipped_unclassified); needs chunks classified into an element schema first.

create_finding

create_finding(
    study_id: str,
    *,
    statement: str,
    supporting_chunk_ids: list[str],
    stance: Stance,
    weight: float,
    agent_id: str,
    finding_type: str = "",
    provenance: Provenance = "primary_text",
    rationale: str = "",
    model: str = "",
    origin_round_id: str = ""
) -> dict[str, Any]

Record a cross-chunk Finding — a pattern asserted over a set of chunks (what per-chunk assess can't see). Same evidence axes as an assessment (stance/weight/provenance) but spanning many chunks; finding_type becomes a routing label. Must cite real chunks. origin_round_id links a finding surfaced by a leveled round.

escalate_study

escalate_study(
    study_id: str,
    *,
    kind: str,
    created_by: str,
    level: int | None = None,
    lens: str | None = None,
    reviewers: int = 1,
    scope: str = "contested",
    limit: int = 50
) -> dict[str, Any]

Open a review round and return only its targeted worklist — more reviewers on contested/low-depth findings (accuracy), or study chunks not yet seen by lens (detectability). Never a blind re-run.

next_review

next_review(
    round_id: str,
    *,
    agent_id: str | None = None,
    limit: int = 20,
    ttl_seconds: int = 1800
) -> list[dict[str, Any]]

Uncovered chunks for a detectability round's lens; with agent_id, atomically claims a non-overlapping batch (punchcard keyed on the round).

record_review

record_review(
    round_id: str,
    target_id: str,
    *,
    target_kind: str = "finding",
    verdict: str | None = None,
    agent_id: str,
    notes: str = "",
    provenance: Provenance | None = None
) -> dict[str, Any]

Record that a round examined a unit (coverage) and, for a finding with a verdict, cast the reviewer vote (updates confidence/escalation_state).

close_round

close_round(round_id: str) -> dict[str, Any]

Close a round (counts the findings it produced; marks it done).

list_rounds

list_rounds(study_id: str) -> list[dict[str, Any]]

A study's review rounds, oldest first (the escalation history).

available_lenses

available_lenses() -> list[dict[str, Any]]

Registered analytical lenses (name + unit_type + description) an escalation can run. Empty until a schema pack registers them.

study_confidence

study_confidence(study_id: str) -> dict[str, Any]

Confidence + named blind spots for a study: per-finding confidence, contested / low-depth worklists, coverage_by_lens (un-run lenses are listed gaps), a recommended next escalation, and whether it's settled.

set_completion_policy

set_completion_policy(
    study_id: str,
    *,
    target_confidence: float = 0.0,
    required_lenses: list[str] | None = None,
    max_rounds: int = 0
) -> dict[str, Any]

Set the bar conclude_study enforces (target confidence, required lenses, max rounds) — makes "done" a checkable contract.

recommend_studies

recommend_studies(study_id: str) -> list[dict[str, Any]]

Propose follow-on studies a study's findings imply (proposals only — never auto-run), each seeded with the triggering findings.

list_recommendations

list_recommendations(study_id: str) -> list[dict[str, Any]]

Follow-on study proposals already recorded for a study.

spawn_study

spawn_study(
    recommendation_id: str, *, approved_by: str
) -> dict[str, Any]

Approve a recommendation → create the child study + SPAWNED_FROM edge.

add_event

add_event(
    doc_id: str,
    *,
    date: str,
    actor: str,
    action: str,
    outcome: str,
    chunk_id: str = "",
    ruling_type: str = "",
    agent_id: str = ""
) -> dict[str, Any]

Record one timeline event (date/actor/action/outcome) on a document, optionally anchored to the chunk it came from.

timeline

timeline(doc_id: str) -> list[dict[str, Any]]

A document's events in chronological order.

timeline_conflicts

timeline_conflicts(doc_id: str) -> dict[str, Any]

Sequence analysis over a document's events: disparate treatment (same trigger → different outcome by actor) + contradictory outcomes. Reports how many events were scanned (honest coverage).

save_report

save_report(
    study_id: str,
    *,
    name: str,
    text: str,
    agent_id: str,
    cites: list[str] | None = None
) -> dict[str, Any]

Save a markdown report on a study (append-only versioned, named) — keep reports on the graph, not as .md litter. Re-saving a name adds a version.

list_reports

list_reports(study_id: str) -> list[dict[str, Any]]

A study's reports — each name with its latest version + version count.

get_report

get_report(
    study_id: str,
    *,
    name: str | None = None,
    version: int | None = None
) -> dict[str, Any] | None

A report's markdown — latest version by default; pass version for a specific one, or omit name for the study's most recent report.

export_report

export_report(
    study_id: str,
    out_path: str,
    *,
    name: str | None = None,
    version: int | None = None
) -> dict[str, Any]

Write a report's markdown to disk on demand (the only time a report becomes a file).

list_findings

list_findings(
    study_id: str, *, finding_type: str | None = None
) -> list[dict[str, Any]]

Cross-chunk findings for a study (weight-ranked), each with its supporting chunks (id + page) and the reviewer-agreement rollup (reviewer_count / vote_tally / agreement / confidence / escalation_state).

verify_finding

verify_finding(
    finding_id: str,
    *,
    verdict: AssessmentVerdict,
    verifier_agent_id: str,
    notes: str = "",
    provenance: Provenance | None = None
) -> dict[str, Any]

A second agent grades a cross-chunk Finding (the independent vote confidence is built from). Self-verification is rejected; recomputes the finding's escalation_state from all votes.

next_unassessed

next_unassessed(
    study_id: str,
    *,
    doc_id: str | None = None,
    section_id: str | None = None,
    element: str | None = None,
    agent_id: str | None = None,
    limit: int = 20,
    ttl_seconds: int = 1800
) -> list[dict[str, Any]]

Work-list of chunks not yet assessed for this study. When agent_id is given, atomically claims (checks out) the returned chunks so parallel analysts don't overlap; without it, a read-only preview. doc_id/section_id scope the work-list (hard filters); element is an advisory scope — chunks classified as that registered element type sort first (the full list is still returned, nothing hidden), so a study reads its subset first without re-scanning. Claims auto-expire after ttl_seconds.

reopen_study

reopen_study(
    study_id: str, *, agent_id: str
) -> dict[str, Any]

Flip a study back to open for deeper analysis.

delete_study

delete_study(study_id: str) -> dict[str, Any]

Cascade-delete a study + its assessments, verification events, and conclusions. Destructive.

cypher

cypher(
    query: str, params: dict[str, Any] | None = None
) -> Any

Run raw Cypher and return kglite's ResultView. It's ergonomic: iterate it (for row in corpus.cypher(...) — each row is a plain dict), index it (result[0]["col"]), take its len(result), list it (result.to_list()), or read result.columns.