diff --git a/CHANGELOG.md b/CHANGELOG.md index bec5914..daeeb4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ ### Added +- **Unified databases (PRO-1618).** `hydradb database create --type unified` provisions a database with ONE corpus instead of separate knowledge and memory corpora, and `database list` shows each database's type. The CLI reads a database's layout once per command from `GET /databases` (`details[].type`) and branches on that, never on a flag: a split database keeps every existing request and rendering exactly as it was, and a unified database never receives `type`. + + On a unified database `query` is a JSON `POST /query` with no `type` and the answer is the four-key unified body (`chunks[]` with `context_id`, `score`, `content`, `enrichment` as a plain string and `enrichment_kind` beside it; `graph[]` with `origin`, `path_summary` and triplets; `forceful_relations[]`, whose `chunk` has the same shape; `llm_prompt`, a markdown document). The human view renders chunks, graph paths grouped by `origin` (query paths apart from chunk relation paths, each chunk relation listed under the returned chunk it hangs under, matched by `relation.chunk_id`) and, when there are any, the forceful relations; the new `--llm` flag prints the server-built markdown `llm_prompt` verbatim on stdout (feedback hint on stderr) so it can be piped into a model call; `--output json` prints the body verbatim, nothing added. `--follow-forceful-relations/--no-follow-forceful-relations` is forwarded. A parser detects the shape by the body (`llm_prompt`, or `graph`/`forceful_relations` as arrays, vs `chunk_content`/`graph_context`; a split body's `graph` and `forceful_relations` are objects), so a unified body that reaches the split path is still rendered as what it is. + + `ingest` on a unified database is a JSON `POST /context/ingest` with the `context` list and one item of exactly one `--text` or `--conversation-file` (a JSON list of `{role, content, name?}` turns), plus `--context-id`, `--title`, `--enrich/--no-enrich`, `--instructions`, `--happened-at` (YYYY-MM-DD), `--attributes` and `--custom-attributes` (JSON objects), `--category`, repeatable `--forceful-relation` and `--acl`, and `--upsert/--no-upsert`. Every value is validated locally and named by turn or flag before a round trip. Files are refused on a unified database with a message pointing at `--text`; `--kind`, `--user-name` and `--markdown` are refused there too, and the unified-only options are refused on a split database. The 202's `results[].source_id` is rendered as the item's context id. + + `list`, `delete`, `relations`, `subgraph` and `inspect` send no `type` on a unified database; an explicit `--kind` is refused there rather than silently dropped, and `delete` keeps its `knowledge` default on a split database. The deprecated aliases that hard-code a kind (`recall full`, `memories add`, `knowledge upload`, ...) are refused on a unified database with the same message. The pinned SDK cannot be relied on for any of this, so the unified calls (and `database create --type`) go over the wrapper's raw v2 path with the same headers, envelope unwrap and error translation as the SDK path. The conformance vectors gain `ingest-unified-json`. + - **`hydradb feedback` — report whether a query's results were actually useful.** `POST /feedback` had no CLI surface. It correlates on one key, the `request_id` from the query's `meta`, and nothing else about the original query is re-sent, so nothing has to be trusted from the client. That key was unreachable: the wrapper's `_unwrap` returns `.data` and drops `meta`, so a successful query discarded its own request id before any caller saw it. `query` now carries it into the payload — additive, since `/query`'s data has no `request_id` of its own, so the documented `--output json` shape gains a key and loses none — and prints it with a copy-pasteable `hydradb feedback` line. It prints on an EMPTY result too: a query that found nothing is the case most worth reporting, and the one with no chunk ids to fall back on. diff --git a/README.md b/README.md index 0692afa..5789c64 100644 --- a/README.md +++ b/README.md @@ -217,7 +217,7 @@ Retrieve knowledge or memories — the single entry point for search. | Option | Description | |--------|-------------| -| `--kind` | Corpus to search: `memory` or `knowledge`. Omit to search both | +| `--kind` | Split databases: corpus to search, `memory` or `knowledge`. Omit to search both. Not used on a unified database | | `--operator` | Keyword operator: `or`, `and`, `phrase` | | `--max-results` / `-n` | Maximum results, 1–50 (default `10`) | | `--mode` / `-m` | Retrieval mode: `fast` or `thinking` | @@ -226,6 +226,8 @@ Retrieve knowledge or memories — the single entry point for search. | `--graph-context` / `--no-graph-context` | Include knowledge graph relations | | `--context` | Additional context to guide retrieval | | `--title` | Exact document title to search inside; repeat the flag for multiple titles | +| `--follow-forceful-relations` / `--no-follow-forceful-relations` | Unified databases: also return chunks pulled in by relations declared at ingest (server default on) | +| `--llm` | Unified databases: print the server-built `llm_prompt` verbatim, ready to inject into a model call | ```bash hydradb query "What did the team say about pricing?" @@ -235,6 +237,26 @@ hydradb query "pricing AND enterprise" --operator and hydradb query "Who owns the rollout?" --title "Q3 Roadmap.md" --title "Smith, John" ``` +On a **unified database** (see `database create --type unified`) the CLI never +sends `type`, and the answer is the four-key unified body: `chunks[]` +(`context_id`, `score`, `content`, `enrichment` as a plain string and +`enrichment_kind` beside it, the declared category, either one omitted when +empty), `graph[]` (`origin`, `path_summary` plus triplets), +`forceful_relations[]` (chunks pulled in by relations declared at ingest, each +`chunk` in the same shape as a `chunks[]` item) and `llm_prompt`, a markdown +document (`# Query results`, `## Results`, `## Forceful relations`, +`## Related facts`, `## Temporal facts`, `## Sources`) with numbered results +to cite. The human view renders the first three, with graph paths grouped by +`origin`: query paths (grown from the query's entities) apart from chunk +relation paths (listed under the returned chunk they hang under). `--llm` +prints the prompt on its own; `--output json` prints the body verbatim. + +```bash +hydradb query "What plan is John on?" --llm | my-model-call +hydradb --output json query "What plan is John on?" | jq '.chunks[].context_id' +hydradb query "refund window" --no-follow-forceful-relations +``` + Every query prints a `request_id`. That is the only key `feedback` correlates on, so keep it if you intend to rate the answer: @@ -277,19 +299,19 @@ round trip. ### ingest -Store a memory, knowledge text, or knowledge file(s). Defaults to `--kind memory`; -file arguments are always knowledge sources. +Store a memory, knowledge text, or knowledge file(s). On a split database it +defaults to `--kind memory`, and file arguments are always knowledge sources. | Option | Description | |--------|-------------| -| `--kind` | `memory` (default) or `knowledge` | +| `--kind` | Split databases: `memory` (default) or `knowledge`. Not used on a unified database | | `--text` / `-t` | Text to ingest. Use `-` to read from stdin | | `--title` | Optional title | -| `--source-id` | Client-assigned source identifier | -| `--user-name` | User name (memory only) | +| `--source-id` | Client-assigned source identifier (the `--context-id` on a unified database) | +| `--user-name` | User name (split memory only) | | `--infer` / `--no-infer` | Extract insights and build the knowledge graph (default on) | -| `--markdown` | Treat text as markdown (memory only) | -| `--upsert` / `--no-upsert` | Update existing items with the same `source_id` (default on) | +| `--markdown` | Treat text as markdown (split memory only) | +| `--upsert` / `--no-upsert` | Update existing items with the same id (default on) | ```bash hydradb ingest --text "User prefers dark mode and weekly email summaries" @@ -302,6 +324,33 @@ echo "piped note" | hydradb ingest `--text`, `--title`, `--source-id`, `--user-name`, `--markdown` and `--no-infer` do not apply to file ingest and are rejected rather than silently ignored. +On a **unified database** `ingest` sends one JSON context item (exactly one of +`--text` or `--conversation-file`) and never a `type`. Files, `--kind`, +`--user-name` and `--markdown` are refused there with a message; these options +apply there and are refused on a split database: + +| Option | Description | +|--------|-------------| +| `--conversation-file` | Path to a JSON list of `{role, content, name?}` turns (roles `user`, `assistant`, `system`) | +| `--context-id` | Caller-assigned id for the item (server-generated when omitted) | +| `--enrich` / `--no-enrich` | Extract facts and graph relations for the item (default on; `--no-infer` means the same) | +| `--instructions` | Steer enrichment for this item | +| `--happened-at` | The event date the item is about, `YYYY-MM-DD` | +| `--attributes` | Declared, filterable attributes as a JSON object | +| `--custom-attributes` | Free-form attributes as a JSON object | +| `--category` | `auto`, `user_preference`, `business_knowledge` or `decision_trace` | +| `--forceful-relation` | A context id this item is declared related to; repeatable | +| `--acl` | A principal allowed to retrieve the item; repeatable | + +```bash +hydradb ingest --text "Refund policy: 30-day window." --context-id policy-1 --title "Refund policy" \ + --happened-at 2026-07-29 --attributes '{"team": "support"}' --category business_knowledge +hydradb ingest --conversation-file ./chat.json --context-id chat-w1 --forceful-relation policy-1 +``` + +The 202 lists each item as `results[].source_id`, which is its context id; +poll it with `hydradb verify `. + --- ### list / inspect / relations / subgraph / verify @@ -310,7 +359,7 @@ Browse and read back what you have stored. | Command | What it does | Key options | |---------|--------------|-------------| -| `list` | Lists ingested sources and memories | `--kind`, `--page`, `--page-size` | +| `list` | Lists ingested sources and memories (`--kind` is for split databases only) | `--kind`, `--page`, `--page-size` | | `inspect ` | Fetches a source's content or a presigned download URL | `--mode` (`content`, `url`, `both`) | | `relations ` | Knowledge graph triplets linked to a source | `--kind`, `--limit` | | `subgraph ` | Everything connected to one item — its thread, replies, parents, children, links — traversed breadth-first | `--kind`, `--depth`, `--max-sources` | @@ -332,8 +381,9 @@ hydradb verify source_abc123 ### delete -Removes memories or knowledge sources by ID. Defaults to `--kind knowledge`, and prompts -for confirmation unless `--yes` is passed. +Removes memories or knowledge sources by ID. On a split database it defaults to +`--kind knowledge`; on a unified database no kind is sent and `--kind` is refused. +Prompts for confirmation unless `--yes` is passed. ```bash hydradb delete source_abc123 --yes @@ -351,8 +401,8 @@ Create and manage databases. | Command | What it does | Key options | |---------|--------------|-------------| -| `database create ` | Provisions a new database | — | -| `database list` | Lists all databases for the authenticated user | — | +| `database create ` | Provisions a new database; `--type unified` gives it one corpus (no `--kind` on later commands) instead of the default `split` layout | `--type` | +| `database list` | Lists all databases for the authenticated user, with each one's type (`split` or `unified`) | - | | `database collections [database]` | Lists collections within a database | — | | `database stats [database]` | Row-count statistics | — | | `database readiness [database]` | Whether the database is ready for ingestion | — | @@ -361,6 +411,7 @@ Create and manage databases. ```bash hydradb database create my-new-database +hydradb database create my-unified-database --type unified hydradb database readiness hydradb database collections hydradb database delete old-database --yes diff --git a/conformance/conftest.py b/conformance/conftest.py index 1fffefc..2220b7e 100644 --- a/conformance/conftest.py +++ b/conformance/conftest.py @@ -18,6 +18,7 @@ from hydra_db import HydraDB as _SdkHydraDB from hydradb_cli.hydra import HydraDB +from hydradb_cli.hydra import client as _client_module VECTORS_PATH = Path(__file__).parent / "vectors.json" @@ -92,9 +93,15 @@ def scope_defaults() -> dict: @pytest.fixture -def wrapper(recorder: Recorder, scope_defaults: dict) -> HydraDB: +def wrapper(recorder: Recorder, scope_defaults: dict, monkeypatch) -> HydraDB: """A wrapper whose SDK talks to the recording mock transport, scoped to the - vectors' default database/collection.""" + vectors' default database/collection. + + The unified calls (PRO-1618) do not go through the SDK: the wrapper sends + them over its raw v2 path with the module-level ``httpx`` functions. Those + are routed to the same recorder, so a unified vector is asserted exactly + the way a split one is. + """ w = HydraDB( token="test-token", base_url="http://conformance.test", @@ -106,4 +113,17 @@ def wrapper(recorder: Recorder, scope_defaults: dict) -> HydraDB: base_url="http://conformance.test", httpx_client=httpx.Client(transport=httpx.MockTransport(recorder.handler)), ) + + def _via_recorder(method: str): + def call(url, *, headers=None, json=None, params=None, timeout=None, **_ignored): + return recorder.handler(httpx.Request(method, url, headers=headers, json=json, params=params)) + + return call + + def _request(method: str, url, **kwargs): + return _via_recorder(method)(url, **kwargs) + + monkeypatch.setattr(_client_module.httpx, "post", _via_recorder("POST")) + monkeypatch.setattr(_client_module.httpx, "get", _via_recorder("GET")) + monkeypatch.setattr(_client_module.httpx, "request", _request) return w diff --git a/conformance/test_conformance.py b/conformance/test_conformance.py index 3c4d151..f944954 100644 --- a/conformance/test_conformance.py +++ b/conformance/test_conformance.py @@ -5,7 +5,8 @@ (a) the wrapper emits the canonical operation (correct endpoint + HTTP method); (b) the SDK call carries the expected fields (``args_include`` / ``args_scope``) - and honours the content-type / forbidden-field guards; + and honours the content-type / forbidden-field guards (a unified-database + call is a raw v2 request rather than an SDK call, recorded the same way); (c) every deprecated **CLI** alias listed resolves to the same canonical operation (same endpoint + method). @@ -49,6 +50,15 @@ def _dispatch(wrapper, op: str, args: dict): if op == "query": return ctx.query(query=args["query"], kind=args.get("kind"), operator=args.get("operator")) if op == "ingest": + if args.get("layout") == "unified": + # PRO-1618: the JSON body with the `context` list, in the contract's + # field names; `id` is the client-assigned context_id. + item = { + key: args[arg] + for arg, key in (("id", "context_id"), ("title", "title"), ("text", "text")) + if args.get(arg) is not None + } + return ctx.ingest_context([item]) return ctx.ingest(kind=args["kind"], text=args.get("text"), title=args.get("title")) if op == "list": return ctx.list(kind=args.get("kind")) diff --git a/conformance/vectors.json b/conformance/vectors.json index b76bbb4..775a16a 100644 --- a/conformance/vectors.json +++ b/conformance/vectors.json @@ -59,6 +59,15 @@ "sdk": { "method": "ingest", "source_field_in": ["app_knowledge"], "item_id_preserved": "claude-file:abc123" } } }, + { + "id": "ingest-unified-json", + "$comment": "PRO-1618. On a UNIFIED database (GET /databases details[].type == \"unified\") ingest is an application/json body whose `context` list carries items of exactly one `text` or one `conversation` each, in the contract's field names (context_id, title, enrich, upsert, instructions, happened_at, attributes, custom_attributes, context_category, forceful_relations, acl). `type` MUST NOT be sent, and neither may any split-era field (documents, app_knowledge, memories, items). The multipart vectors above are unchanged for split databases: a client branches on the database's layout, never on a request flag. `layout` in `args` tells the runner which database the call is aimed at.", + "call": { "op": "ingest", "args": { "layout": "unified", "text": "quarterly report body", "title": "Q3", "id": "src_client_assigned" } }, + "expect": { + "wrapper_method": "context.ingestContext", + "sdk": { "method": "raw", "content_type": "application/json", "args_include": { "database": "db_test", "context": [ { "context_id": "src_client_assigned", "title": "Q3", "text": "quarterly report body" } ] }, "args_scope": { "database": "db_test", "collection": "col_test" }, "forbid_content_type": "multipart/form-data", "forbid_field": "type", "source_field_in": ["context"] } + } + }, { "id": "list-memory", "call": { "op": "list", "args": { "kind": "memory" } }, diff --git a/src/hydradb_cli/commands/_impl.py b/src/hydradb_cli/commands/_impl.py index b1a97b3..18ddd00 100644 --- a/src/hydradb_cli/commands/_impl.py +++ b/src/hydradb_cli/commands/_impl.py @@ -10,11 +10,15 @@ from __future__ import annotations +import json +import re from collections.abc import Callable +from datetime import date from pathlib import Path from typing import Any import httpx +import typer from rich.console import Group from rich.markup import escape from rich.panel import Panel @@ -22,7 +26,18 @@ from rich.text import Text from hydradb_cli.hydra import HydraDBClientError -from hydradb_cli.output import make_kv_table, make_table, print_error, print_result, spinner +from hydradb_cli.hydra.client import LAYOUT_SPLIT, LAYOUT_UNIFIED +from hydradb_cli.output import ( + console, + err_console, + get_output_format, + make_kv_table, + make_table, + print_error, + print_json, + print_result, + spinner, +) from hydradb_cli.utils.common import ( get_wrapper, handle_api_error, @@ -41,6 +56,12 @@ # "agnet" is worth catching before it costs one. VALID_SOURCES = {"user", "agent"} VALID_FETCH_MODES = {"content", "url", "both"} +# Unified ingest (PRO-1618): the context_category labels and conversation roles +# the server accepts. Validated locally for the same reason --rating is. +VALID_CATEGORIES = {"auto", "user_preference", "business_knowledge", "decision_trace"} +VALID_ROLES = {"user", "assistant", "system"} +# happened_at is a calendar date, YYYY-MM-DD only: no time, no zone. +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") _STATUS_LABELS = { "queued": "queued", @@ -71,6 +92,61 @@ def _execute(spinner_msg: str, call: Callable[[], Any]) -> Any: handle_network_error(e) +# ── storage layout (PRO-1618) ──────────────────────────────────────────────── + + +def _is_unified(wrapper: Any, database: str) -> bool: + """Whether ``database`` is a unified database. + + One memoised ``GET /databases`` probe per wrapper. A successful probe that + does not list ``database`` reads as split, which is what every pre-PRO-1618 + database is; a FAILED probe is the error it is, not a guess: guessing + split would send the split request shape to a database that may be + unified. Compared by value so a mocked wrapper (whose ``layout`` returns a + MagicMock) reads as split too. + Every command branches on THIS, never on a request flag: a unified database + never receives ``type``, and a split one keeps every existing call as is. + """ + try: + return wrapper.databases.layout(database) == LAYOUT_UNIFIED + except HydraDBClientError as e: + handle_api_error(e) + except httpx.RequestError as e: + handle_network_error(e) + + +def database_layout(tenant_id: str | None) -> tuple[str, str]: + """The database a command is about to touch and its layout, ``unified`` or ``split``.""" + tid = require_tenant_id(tenant_id) + return tid, (LAYOUT_UNIFIED if _is_unified(get_wrapper(), tid) else LAYOUT_SPLIT) + + +def _refuse_kind_on_unified(kind: str | None, database: str) -> None: + """A unified database has one corpus, so there is no kind to select. + + The server refuses ``knowledge``/``memory`` there and the contract says + never to send ``type`` at all. Refused locally, naming the rule, rather + than silently swapped for something the user did not ask for. + """ + if kind: + print_error( + f"Database '{database}' is unified: it has one corpus, so a kind ('{kind}') cannot be selected. " + "Re-run without --kind." + ) + + +def _refuse_split_write_on_unified(wrapper: Any, database: str, layout: str | None, what: str) -> None: + """The split ingest shapes (``memories``/``app_knowledge``/``documents`` + with ``type``) are refused by a unified database. Probed here unless the + caller already resolved the layout, so the deprecated aliases are covered.""" + unified = layout == LAYOUT_UNIFIED if layout is not None else _is_unified(wrapper, database) + if unified: + print_error( + f"Database '{database}' is unified: {what}. " + "Use 'hydradb ingest --text ...' or 'hydradb ingest --conversation-file ...' (no --kind)." + ) + + # ── query ──────────────────────────────────────────────────────────────────── @@ -87,7 +163,233 @@ def _feedback_hint(r: dict) -> str: return f'\n[dim]request_id: {request_id} · rate it: hydradb feedback {request_id} --feedback "..."[/dim]' +def _is_unified_query_body(r: dict) -> bool: + """Shape detection (contract rule 4). + + A unified body carries ``llm_prompt``, a ``graph`` ARRAY and a + ``forceful_relations`` ARRAY; a split body carries ``chunk_content``/ + ``graph_context``. A split body can carry ``graph`` and + ``forceful_relations`` too, but as objects (``{paths}``, + ``{declared, inferred}``), so both are told apart by type, never by the key + being there. Stored logs and split databases keep producing the old shape, + so the layout probe alone cannot decide this. + """ + return "llm_prompt" in r or isinstance(r.get("graph"), list) or isinstance(r.get("forceful_relations"), list) + + +def _preview(text: str, limit: int) -> str: + return text[:limit] + "..." if len(text) > limit else text + + +def _pct(score: Any) -> str: + return f"{score:.0%}" if isinstance(score, (int, float)) and not isinstance(score, bool) else "" + + +def _unified_chunk_panel(chunk: dict, label: str) -> Panel: + """One ``chunks[]`` item: context_id, score, content, enrichment and + enrichment_kind, temporal facts. ``enrichment`` is a plain string and + ``enrichment_kind`` its sibling (the declared context_category); either + can be absent, and a kind with no enrichment is still shown. Content and + enrichment are shown whole, never trimmed (the unified answer is not + compacted anywhere). Content is API data, so it is rendered as plain Text + and never parsed as markup.""" + score = _pct(chunk.get("score")) + score_str = f" • {score}" if score else "" + context_id = chunk.get("context_id") or "" + id_str = f" • {escape(str(context_id))}" if context_id else "" + body: list[Any] = [Text(chunk.get("content") or "")] + enrichment = chunk.get("enrichment") + enrichment = enrichment if isinstance(enrichment, str) else "" + kind = chunk.get("enrichment_kind") + kind = kind if isinstance(kind, str) else "" + if enrichment or kind: + head = f"enrichment ({kind})" if kind else "enrichment" + body.append(Text.assemble((f"{head}: " if enrichment else head, "dim"), enrichment)) + for fact in chunk.get("temporal") or []: + if isinstance(fact, dict): + span = " to ".join(str(x) for x in (fact.get("start_date"), fact.get("end_date")) if x) + body.append( + Text.assemble(("temporal: ", "dim"), fact.get("content") or "", (f" [{span}]" if span else "", "dim")) + ) + return Panel( + Group(*body), + title=f"[bold]{label}[/bold]{score_str}{id_str}", + title_align="left", + border_style="cyan", + padding=(0, 1), + ) + + +#: ``graph[].origin`` (PRO-1618): which retrieval lane found a path. +#: ``query_path`` was grown from the entities in the query; ``chunk_relation`` +#: is the neighbourhood of a chunk that ranked. +_ORIGIN_QUERY_PATH = "query_path" +_ORIGIN_CHUNK_RELATION = "chunk_relation" + + +def _list_field(r: dict, key: str) -> list: + """``r[key]`` when it is a list, else ``[]``. The unified keys are read by + type, never by presence: a split body has a ``graph`` and a + ``forceful_relations`` too, but as objects.""" + value = r.get(key) + return value if isinstance(value, list) else [] + + +def _chunk_labels(chunks: list, forceful: list) -> dict[str, tuple[str, str]]: + """``chunk_id -> (label, context_id)`` for every chunk the body returned: + ``1``, ``2``... for ``chunks[]`` and ``R1``, ``R2``... for + ``forceful_relations[].chunk``, the labels ``llm_prompt`` cites them by. + First write wins, as on the server: a chunk that is both a result and a + forceful relation is cited as the result. This is how a graph hop is tied + to its chunk and context: by ``relation.chunk_id``, never by parsing it.""" + labels: dict[str, tuple[str, str]] = {} + entries = [(str(i), chunk) for i, chunk in enumerate(chunks, 1)] + entries += [(f"R{i}", rel.get("chunk") or {}) for i, rel in enumerate(forceful, 1)] + for label, chunk in entries: + chunk_id = chunk.get("chunk_id") + if chunk_id and chunk_id not in labels: + labels[chunk_id] = (label, chunk.get("context_id") or "") + return labels + + +def _hop_chunk(triplet: dict, labels: dict[str, tuple[str, str]]) -> tuple[str, str] | None: + return labels.get((triplet.get("relation") or {}).get("chunk_id") or "") + + +def _hop_line(triplet: dict, labels: dict[str, tuple[str, str]], *, cite: bool) -> str: + src = (triplet.get("source") or {}).get("name") or "?" + predicate = (triplet.get("relation") or {}).get("predicate") or "related to" + tgt = (triplet.get("target") or {}).get("name") or "?" + line = f"{src} -> {predicate} -> {tgt}" + hit = _hop_chunk(triplet, labels) if cite else None + return f"{line} [{hit[0]}]" if hit else line + + +def _graph_panels(graph: list, labels: dict[str, tuple[str, str]]) -> list[Panel]: + """``graph[]`` grouped by ``origin``, the way the split body kept + ``query_paths`` and ``chunk_relations`` apart. + + Query paths are listed with each hop citing the returned chunk it was + extracted from, when there is one. Chunk relations are listed under the + chunk they hang under: every hop's ``relation.chunk_id`` is that chunk. + A path with no known origin is still shown, in a group of its own, rather + than guessed into one. ``P`` labels are positions in ``graph[]``, which is + what ``llm_prompt`` numbers paths by. + """ + query_rows: list[list[str]] = [] + chunk_rows: list[list[str]] = [] + other_rows: list[list[str]] = [] + for i, path in enumerate(graph, 1): + if not isinstance(path, dict): + continue + triplets = [t for t in path.get("triplets") or [] if isinstance(t, dict)] + summary = path.get("path_summary") or "" + origin = path.get("origin") + if origin == _ORIGIN_CHUNK_RELATION: + under: list[str] = [] + for triplet in triplets: + hit = _hop_chunk(triplet, labels) + cell = f"[{hit[0]}] {hit[1]}".rstrip() if hit else "" + if cell and cell not in under: + under.append(cell) + hops = "\n".join(_hop_line(t, labels, cite=False) for t in triplets) + chunk_rows.append([f"P{i}", "\n".join(under), summary, hops]) + continue + hops = "\n".join(_hop_line(t, labels, cite=True) for t in triplets) + (query_rows if origin == _ORIGIN_QUERY_PATH else other_rows).append([f"P{i}", summary, hops]) + + groups = ( + (query_rows, "query path(s)", ("#", "Path", "Triplets")), + (chunk_rows, "chunk relation path(s)", ("#", "Chunk", "Path", "Triplets")), + (other_rows, "path(s) with no known origin", ("#", "Path", "Triplets")), + ) + return [ + Panel( + make_table(*columns, rows=rows), + title=f"[bold cyan]/// Graph: {len(rows)} {what}[/bold cyan]", + border_style="cyan", + padding=(0, 1), + ) + for rows, what, columns in groups + if rows + ] + + +def _forceful_relations_panel(forceful: list) -> Panel: + """``forceful_relations[]``: chunks in the result because the caller + declared a relation at ingest, with the declared edge (``via``) that + pulled each one in. ``R`` labels match ``llm_prompt``'s.""" + rows = [] + for i, rel in enumerate(forceful, 1): + via = rel.get("via") or {} + chunk = rel.get("chunk") or {} + rows.append( + [ + f"R{i}", + via.get("from") or "", + via.get("to") or chunk.get("context_id") or "", + _pct(chunk.get("score")), + chunk.get("content") or "", + ] + ) + return Panel( + make_table("#", "From", "To", "Score", "Content", rows=rows), + title=f"[bold cyan]/// Forceful relations: {len(forceful)} chunk(s)[/bold cyan]", + border_style="cyan", + padding=(0, 1), + ) + + +def _format_unified_query_result(r: dict, request_id: str | None = None): + """The four-key unified body (PRO-1618) for a person: ``chunks[]`` as + panels, ``graph[]`` as path tables grouped by ``origin``, + ``forceful_relations[]`` as a table of what declared relations pulled in + (no section when there are none), and the feedback hint. ``llm_prompt`` is + not shown here: ``--llm`` prints it verbatim.""" + chunks = _list_field(r, "chunks") + graph = _list_field(r, "graph") + forceful = [f for f in _list_field(r, "forceful_relations") if isinstance(f, dict)] + hint = _feedback_hint({"request_id": request_id}) if request_id else "" + if not chunks and not graph and not forceful: + return "[dim]No relevant results found.[/dim]" + hint + + parts: list[Any] = [Text(f" Found {len(chunks)} result(s)", style="bold")] + for i, chunk in enumerate(chunks, 1): + parts.append(_unified_chunk_panel(chunk, str(i))) + + parts.extend(_graph_panels(graph, _chunk_labels(chunks, forceful))) + + if forceful: + parts.append(_forceful_relations_panel(forceful)) + + if hint: + parts.append(Text.from_markup(hint.lstrip("\n"))) + return Group(*parts) + + +def _print_unified_query(body: dict, request_id: str | None, *, llm: bool) -> None: + """Print a unified query result. ``--output json`` is the body verbatim, the + four keys and nothing added; ``--llm`` is the server-built prompt on plain + stdout (not Rich, which would re-wrap it at the terminal width and read + bracketed fragments as markup), with the feedback hint on stderr so the + prompt can be piped; otherwise the structured rendering.""" + if get_output_format() == "json": + print_json(body) + return + if llm: + typer.echo(body.get("llm_prompt") or "") + if request_id: + err_console.print(_feedback_hint({"request_id": request_id}).lstrip("\n")) + return + console.print(_format_unified_query_result(body, request_id)) + + def _format_query_result(r: dict): + if _is_unified_query_body(r): + # A unified body reached the split path (a server that does not list a + # database's layout still answers a type-less query on a unified one + # in the unified shape). Render it as what it is. + return _format_unified_query_result(r, r.get("request_id")) chunks = r.get("chunks") or [] if not chunks: return "[dim]No relevant results found.[/dim]" + _feedback_hint(r) @@ -138,6 +440,8 @@ def do_query( additional_context: str | None = None, titles: list[str] | None = None, acl: list[str] | None = None, + follow_forceful_relations: bool | None = None, + llm: bool = False, tenant_id: str | None = None, sub_tenant_id: str | None = None, spinner_msg: str = "Searching...", @@ -173,6 +477,39 @@ def do_query( stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + if _is_unified(wrapper, tid): + _refuse_kind_on_unified(kind, tid) + outcome = _execute( + spinner_msg, + lambda: wrapper.context.query_unified( + query=query, + operator=operator, + query_by="text" if operator else None, + max_results=max_results, + mode=mode, + alpha=alpha, + recency_bias=recency_bias, + graph_context=graph_context, + additional_context=additional_context, + titles=clean_titles, + acl=acl, + follow_forceful_relations=follow_forceful_relations, + database=tid, + collection=stid, + ), + ) + body, request_id = outcome if isinstance(outcome, tuple) else (outcome, None) + _print_unified_query(body if isinstance(body, dict) else {}, request_id, llm=llm) + return + + if follow_forceful_relations is not None: + print_error( + "--follow-forceful-relations/--no-follow-forceful-relations applies to unified databases only; " + f"'{tid}' is a split database." + ) + if llm: + print_error(f"--llm applies to unified databases only; '{tid}' is a split database and has no llm_prompt.") + result = _execute( spinner_msg, lambda: wrapper.context.query( @@ -302,10 +639,12 @@ def do_ingest_memory( upsert: bool = True, tenant_id: str | None = None, sub_tenant_id: str | None = None, + layout: str | None = None, ) -> None: tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + _refuse_split_write_on_unified(wrapper, tid, layout, "a memory (kind) cannot be written to it") result = _execute( "Adding memory...", @@ -332,10 +671,12 @@ def do_ingest_knowledge_text( source_id: str | None = None, tenant_id: str | None = None, sub_tenant_id: str | None = None, + layout: str | None = None, ) -> None: tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + _refuse_split_write_on_unified(wrapper, tid, layout, "knowledge text (kind) cannot be written to it") result = _execute( "Uploading text...", @@ -385,6 +726,7 @@ def do_ingest_knowledge_files( upsert: bool = False, tenant_id: str | None = None, sub_tenant_id: str | None = None, + layout: str | None = None, ) -> None: if not files: print_error("At least one file path is required.") @@ -405,6 +747,9 @@ def do_ingest_knowledge_files( tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + _refuse_split_write_on_unified( + wrapper, tid, layout, "files are not accepted (text or a conversation only). Extract the text first" + ) result = _execute( f"Uploading {len(files)} file(s)...", @@ -444,6 +789,211 @@ def fmt(r: dict): print_result(result, fmt) +# ── ingest (unified databases, PRO-1618) ───────────────────────────────────── + + +def _load_conversation(path: str) -> list[dict[str, Any]]: + """Read ``--conversation-file``: a JSON list of ``{role, content, name?}`` + turns. Every turn is checked here so a bad one is named by index locally + rather than as ``context[0]`` after a round trip.""" + p = Path(path) + if not p.is_file(): + print_error(f"Conversation file not found: {path}") + try: + turns = json.loads(p.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + print_error(f"--conversation-file must be a JSON list of {{role, content, name?}} turns: {exc}") + if not isinstance(turns, list) or not turns: + print_error("--conversation-file must be a non-empty JSON list of {role, content, name?} turns.") + clean: list[dict[str, Any]] = [] + for i, turn in enumerate(turns): + if not isinstance(turn, dict): + print_error(f"conversation[{i}] must be an object with role and content.") + role = turn.get("role") + if role not in VALID_ROLES: + print_error(f"conversation[{i}].role must be one of: {', '.join(sorted(VALID_ROLES))}. Got {role!r}.") + content = turn.get("content") + if not isinstance(content, str) or not content.strip(): + print_error(f"conversation[{i}].content must be a non-empty string.") + unknown = sorted(set(turn) - {"role", "content", "name"}) + if unknown: + print_error( + f"conversation[{i}] has unknown field(s): {', '.join(unknown)}. Only role, content and name are accepted." + ) + item: dict[str, Any] = {"role": role, "content": content} + name = turn.get("name") + if name is not None: + if not isinstance(name, str) or not name.strip(): + print_error(f"conversation[{i}].name must be a non-empty string when present.") + item["name"] = name + clean.append(item) + return clean + + +def _parse_json_object(raw: str | None, flag: str) -> dict[str, Any] | None: + if raw is None: + return None + try: + parsed = json.loads(raw) + except ValueError as exc: + print_error(f"{flag} must be a JSON object: {exc}") + if not isinstance(parsed, dict): + print_error(f'{flag} must be a JSON object, for example \'{{"team": "support"}}\'.') + return parsed + + +def build_context_item( + *, + text: str | None = None, + conversation: list[dict[str, Any]] | None = None, + context_id: str | None = None, + title: str | None = None, + enrich: bool = True, + instructions: str | None = None, + happened_at: str | None = None, + attributes: dict[str, Any] | None = None, + custom_attributes: dict[str, Any] | None = None, + category: str | None = None, + forceful_relations: list[str] | None = None, + acl: list[str] | None = None, + upsert: bool = True, +) -> dict[str, Any]: + """One ``context[]`` item in the contract's exact field names (PRO-1618). + + Exactly one of ``text`` or ``conversation``. Optional fields are omitted + when unset, never sent as null; ``enrich`` and ``upsert`` are always sent + because the CLI always has a value for them. + """ + if (text is None) == (conversation is None): + print_error("Pass exactly one of --text or --conversation-file.") + item: dict[str, Any] = {} + if context_id: + item["context_id"] = context_id + if title: + item["title"] = title + if text is not None: + item["text"] = text + else: + item["conversation"] = conversation + item["enrich"] = bool(enrich) + item["upsert"] = bool(upsert) + if instructions: + item["instructions"] = instructions + if happened_at: + try: + valid = bool(_DATE_RE.match(happened_at)) and date.fromisoformat(happened_at) is not None + except ValueError: + valid = False + if not valid: + print_error(f"--happened-at must be a calendar date in YYYY-MM-DD form, got '{happened_at}'.") + item["happened_at"] = happened_at + if attributes is not None: + item["attributes"] = attributes + if custom_attributes is not None: + item["custom_attributes"] = custom_attributes + if category: + if category not in VALID_CATEGORIES: + print_error(f"--category must be one of: {', '.join(sorted(VALID_CATEGORIES))}. Got '{category}'.") + item["context_category"] = category + if forceful_relations: + ids: list[str] = [] + for value in forceful_relations: + candidate = (value or "").strip() + if not candidate: + print_error("--forceful-relation cannot be empty or whitespace-only.") + if candidate not in ids: + ids.append(candidate) + item["forceful_relations"] = {"ids": ids} + if acl is not None: + item["acl"] = list(acl) + return item + + +def _format_ingest_unified(r: dict, item: dict[str, Any]): + success_count = r.get("success_count", 0) + failed_count = r.get("failed_count", 0) + ok = failed_count == 0 + status = "green" if ok else "yellow" + mark = "✓" if ok else "!" + if "conversation" in item: + preview = f"conversation, {len(item['conversation'])} turn(s)" + else: + preview = f'"{_preview(item.get("text") or "", 80)}"' + lines = [ + f"[{status}]{mark}[/{status}] Context queued ({success_count} success, {failed_count} failed)", + f"[dim]{escape(preview)}[/dim]", + ] + for res in r.get("results", []) or []: + # The 202 still spells the item's context_id `source_id`. + cid = res.get("source_id") or res.get("context_id") or res.get("id") or "unknown" + lines.append(f"[cyan]Context ID:[/cyan] {escape(str(cid))} [dim]({res.get('status', 'unknown')})[/dim]") + if res.get("error"): + code = f" ({res['error_code']})" if res.get("error_code") else "" + lines.append(f"[red]Error:[/red] {escape(str(res['error']))}{escape(code)}") + lines.append("[dim]Poll with 'hydradb verify '.[/dim]") + return Panel("\n".join(lines), border_style=status, padding=(0, 1)) + + +def do_ingest_unified( + *, + text: str | None = None, + conversation_file: str | None = None, + context_id: str | None = None, + title: str | None = None, + enrich: bool = True, + instructions: str | None = None, + happened_at: str | None = None, + attributes: str | None = None, + custom_attributes: str | None = None, + category: str | None = None, + forceful_relations: list[str] | None = None, + acl: list[str] | None = None, + upsert: bool = True, + tenant_id: str | None = None, + sub_tenant_id: str | None = None, +) -> None: + """Ingest one context item into a UNIFIED database: a JSON ``POST + /context/ingest`` with the ``context`` list, no ``type``, no multipart.""" + tid = require_tenant_id(tenant_id) + stid = resolve_sub_tenant_id(sub_tenant_id) + if conversation_file and text is not None: + print_error("Pass exactly one of --text or --conversation-file.") + conversation = _load_conversation(conversation_file) if conversation_file else None + item = build_context_item( + text=text, + conversation=conversation, + context_id=context_id, + title=title, + enrich=enrich, + instructions=instructions, + happened_at=happened_at, + attributes=_parse_json_object(attributes, "--attributes"), + custom_attributes=_parse_json_object(custom_attributes, "--custom-attributes"), + category=category, + forceful_relations=forceful_relations, + acl=acl, + upsert=upsert, + ) + wrapper = get_wrapper() + + result = _execute( + "Ingesting context...", + lambda: wrapper.context.ingest_context([item], database=tid, collection=stid), + ) + print_result(result, lambda r: _format_ingest_unified(r, item)) + # A 202 only means accepted: the per-item verdicts are in the body, and a + # failure among them is a failed ingest. The panel above already printed + # the server's own error for the row; the exit code is what says the + # command did not succeed. + failed_rows = [ + r + for r in (result.get("results") or []) + if isinstance(r, dict) and (r.get("error") or r.get("status") in ("failed", "errored")) + ] + if (result.get("failed_count") or 0) > 0 or failed_rows: + raise typer.Exit(code=1) + + # ── list ───────────────────────────────────────────────────────────────────── @@ -467,6 +1017,9 @@ def do_list( tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + if _is_unified(wrapper, tid): + # One corpus: no kind is selected and none is sent. + _refuse_kind_on_unified(kind, tid) result = _execute( spinner_msg, @@ -577,26 +1130,33 @@ def fmt(r: dict): def do_delete( ids: list[str], *, - kind: str, + kind: str | None, tenant_id: str | None = None, sub_tenant_id: str | None = None, ) -> None: clean_ids = [i.strip() for i in ids if i.strip()] if not clean_ids: print_error("IDs cannot be empty.") - if kind not in VALID_KINDS: + if kind is not None and kind not in VALID_KINDS: print_error(f"--kind must be one of: {', '.join(sorted(VALID_KINDS))}. Got '{kind}'.") tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + if _is_unified(wrapper, tid): + # One corpus: no kind is selected and none is sent. + _refuse_kind_on_unified(kind, tid) + noun = "item(s)" + else: + # The split default, unchanged: a delete without --kind is a knowledge delete. + kind = kind or "knowledge" + noun = "memory" if kind == "memory" else "knowledge source(s)" result = _execute( "Deleting...", lambda: wrapper.context.delete(ids=clean_ids, kind=kind, database=tid, collection=stid), ) - noun = "memory" if kind == "memory" else "knowledge source(s)" # v2 returns HTTP 200 with {success:false, deleted_count:0} when nothing # matched — that is a no-op, not a success. Surface it as an error (non-zero # exit, and `{"success":false,"error":…}` in json mode) rather than claiming @@ -631,6 +1191,9 @@ def do_relations( tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + if _is_unified(wrapper, tid): + # One corpus: no kind is selected and none is sent. + _refuse_kind_on_unified(kind, tid) result = _execute( "Fetching graph relations...", @@ -689,6 +1252,9 @@ def do_subgraph( tid = require_tenant_id(tenant_id) stid = resolve_sub_tenant_id(sub_tenant_id) wrapper = get_wrapper() + if _is_unified(wrapper, tid): + # One corpus: no kind is selected and none is sent. + _refuse_kind_on_unified(kind, tid) result = _execute( "Traversing the connected subgraph...", @@ -804,9 +1370,11 @@ def fmt(r: dict): # ── database group ─────────────────────────────────────────────────────────── -def do_database_create(database: str) -> None: +def do_database_create(database: str, layout: str | None = None) -> None: if not database.strip(): print_error("Database ID cannot be empty.") + if layout and layout not in (LAYOUT_SPLIT, LAYOUT_UNIFIED): + print_error(f"--type must be '{LAYOUT_SPLIT}' or '{LAYOUT_UNIFIED}'. Got '{layout}'.") # is_embeddings_tenant is deliberately not passed. The API treats it as an # internal flag: it provisions a raw-embeddings collection *instead of* the @@ -815,9 +1383,10 @@ def do_database_create(database: str) -> None: wrapper = get_wrapper() result = _execute( "Creating database...", - lambda: wrapper.databases.create(database=database), + lambda: wrapper.databases.create(database=database, layout=layout), ) - print_result(result, lambda r: f"[green]✓[/green] Database [bold]{database}[/bold] created successfully.") + suffix = " (unified: one corpus, no --kind on later commands)" if layout == LAYOUT_UNIFIED else "" + print_result(result, lambda r: f"[green]✓[/green] Database [bold]{database}[/bold] created successfully.{suffix}") def do_database_delete(database: str) -> None: @@ -836,7 +1405,11 @@ def fmt(r: dict): ids = r.get("databases") or r.get("tenant_ids") or [] if not ids: return "[dim]No databases found.[/dim]" - return make_table("Database ID", rows=[[i] for i in ids], title=f"Found {len(ids)} database(s)") + # `details[]` (PRO-1618) carries each database's storage layout; a + # server that omits it has only split databases. + layouts = {row.get("database"): row.get("type") for row in (r.get("details") or []) if isinstance(row, dict)} + rows = [[i, LAYOUT_UNIFIED if layouts.get(i) == LAYOUT_UNIFIED else LAYOUT_SPLIT] for i in ids] + return make_table("Database ID", "Type", rows=rows, title=f"Found {len(ids)} database(s)") print_result(result, fmt) diff --git a/src/hydradb_cli/commands/canonical.py b/src/hydradb_cli/commands/canonical.py index 909a010..c4d722c 100644 --- a/src/hydradb_cli/commands/canonical.py +++ b/src/hydradb_cli/commands/canonical.py @@ -47,7 +47,9 @@ def _resolve_text_input(text: str | None) -> str: def query( query_text: str = typer.Argument(metavar="QUERY", help="Search query."), - kind: str | None = typer.Option(None, "--kind", help="Corpus to query: 'memory' or 'knowledge'."), + kind: str | None = typer.Option( + None, "--kind", help="Corpus to query on a split database: 'memory' or 'knowledge'. Not used on a unified one." + ), operator: str | None = typer.Option(None, "--operator", help="Keyword operator: 'or', 'and', or 'phrase'."), max_results: int = typer.Option(10, "--max-results", "-n", help="Maximum number of results (1-50)."), mode: str | None = typer.Option(None, "--mode", "-m", help="Retrieval mode: 'fast' or 'thinking'."), @@ -67,6 +69,16 @@ def query( "--acl", help="Principals to answer as, repeatable (--acl alice@corp.com --acl 'group:google:eng@corp.com'). Restricts results to documents whose access list admits one of them. Omit to search everything the API key can reach.", ), + follow_forceful_relations: bool | None = typer.Option( + None, + "--follow-forceful-relations/--no-follow-forceful-relations", + help="Unified databases only: also return chunks pulled in by relations declared at ingest (server default on).", + ), + llm: bool = typer.Option( + False, + "--llm", + help="Unified databases only: print the server-built llm_prompt verbatim, ready to inject into a model call.", + ), database: str | None = typer.Option(None, "--database", "-d", help="Database. Uses default if not specified."), collection: str | None = typer.Option(None, "--collection", help="Collection."), tenant_id: str | None = typer.Option(None, "--tenant-id", hidden=True), @@ -86,6 +98,8 @@ def query( additional_context=additional_context, titles=list(titles) if titles else None, acl=list(acl) if acl else None, + follow_forceful_relations=follow_forceful_relations, + llm=llm, tenant_id=tid, sub_tenant_id=stid, ) @@ -128,22 +142,126 @@ def feedback( def ingest( - files: list[str] | None = typer.Argument(None, help="Knowledge file path(s) to ingest."), - kind: str | None = typer.Option(None, "--kind", help="Kind to ingest: 'memory' (default) or 'knowledge'."), + files: list[str] | None = typer.Argument(None, help="Knowledge file path(s) to ingest (split databases only)."), + kind: str | None = typer.Option( + None, + "--kind", + help="Kind to ingest on a split database: 'memory' (default) or 'knowledge'. Not used on a unified one.", + ), text: str | None = typer.Option(None, "--text", "-t", help="Text to ingest. Use '-' to read from stdin."), title: str | None = typer.Option(None, "--title", help="Optional title."), - source_id: str | None = typer.Option(None, "--source-id", help="Source identifier."), - user_name: str | None = typer.Option(None, "--user-name", help="User name (memory only)."), + source_id: str | None = typer.Option( + None, "--source-id", help="Source identifier (--context-id on a unified database)." + ), + user_name: str | None = typer.Option(None, "--user-name", help="User name (split memory only)."), infer: bool = typer.Option(True, "--infer/--no-infer", help="Extract insights and build knowledge graph."), - markdown: bool = typer.Option(False, "--markdown", help="Treat text as markdown (memory only)."), - upsert: bool = typer.Option(True, "--upsert/--no-upsert", help="Update existing items with the same source_id."), + markdown: bool = typer.Option(False, "--markdown", help="Treat text as markdown (split memory only)."), + upsert: bool = typer.Option(True, "--upsert/--no-upsert", help="Update existing items with the same id."), + conversation_file: str | None = typer.Option( + None, + "--conversation-file", + help="Unified databases: path to a JSON list of {role, content, name?} turns to ingest as one conversation (roles: user, assistant, system).", + ), + context_id: str | None = typer.Option( + None, + "--context-id", + help="Unified databases: caller-assigned id for the item (server-generated when omitted). --source-id means the same there.", + ), + enrich: bool = typer.Option( + True, + "--enrich/--no-enrich", + help="Unified databases: extract facts and graph relations for the item. --no-infer means the same there.", + ), + instructions: str | None = typer.Option( + None, "--instructions", help="Unified databases: steer enrichment for this item." + ), + happened_at: str | None = typer.Option( + None, "--happened-at", help="Unified databases: the event date the item is about, YYYY-MM-DD." + ), + attributes: str | None = typer.Option( + None, "--attributes", help="Unified databases: declared, filterable attributes as a JSON object." + ), + custom_attributes: str | None = typer.Option( + None, "--custom-attributes", help="Unified databases: free-form attributes as a JSON object." + ), + category: str | None = typer.Option( + None, + "--category", + help="Unified databases: context_category label: 'auto', 'user_preference', 'business_knowledge' or 'decision_trace'.", + ), + forceful_relation: list[str] | None = typer.Option( + None, + "--forceful-relation", + help="Unified databases: a context id this item is declared related to; repeatable.", + ), + acl: list[str] | None = typer.Option( + None, + "--acl", + help="Unified databases: a principal allowed to retrieve the item, repeatable (user_email:a@x.com, domain:acme.com).", + ), database: str | None = typer.Option(None, "--database", "-d", help="Database. Uses default if not specified."), collection: str | None = typer.Option(None, "--collection", help="Collection."), tenant_id: str | None = typer.Option(None, "--tenant-id", hidden=True), sub_tenant_id: str | None = typer.Option(None, "--sub-tenant-id", hidden=True), ) -> None: - """Ingest a memory, knowledge text, or knowledge file(s).""" + """Ingest a memory, knowledge text, or knowledge file(s); on a unified database, one context item.""" tid, stid = resolve_scope_flags(database, collection, tenant_id, sub_tenant_id) + # The layout decides the request shape (PRO-1618), never a flag: a unified + # database gets one JSON context item and never a kind; a split one keeps + # every existing call exactly as it was. + db, layout = _impl.database_layout(tid) + if layout == "unified": + if files: + print_error( + f"Database '{db}' is unified: files are not accepted (text or a conversation only). " + "Extract the text and pass it with --text." + ) + if kind: + print_error(f"Database '{db}' is unified: it has one corpus, so --kind does not apply. Omit it.") + if user_name: + print_error( + "--user-name does not apply on a unified database; name speakers per turn in --conversation-file." + ) + if markdown: + print_error("--markdown does not apply on a unified database.") + if conversation_file and text: + print_error("Pass exactly one of --text or --conversation-file.") + if context_id and source_id and context_id != source_id: + print_error("--context-id and --source-id name the same thing on a unified database; pass one of them.") + _impl.do_ingest_unified( + text=None if conversation_file else _resolve_text_input(text), + conversation_file=conversation_file, + context_id=context_id or source_id, + title=title, + enrich=enrich and infer, + instructions=instructions, + happened_at=happened_at, + attributes=attributes, + custom_attributes=custom_attributes, + category=category, + forceful_relations=list(forceful_relation) if forceful_relation else None, + acl=list(acl) if acl else None, + upsert=upsert, + tenant_id=tid, + sub_tenant_id=stid, + ) + return + + unified_only = { + "--conversation-file": conversation_file, + "--context-id": context_id, + "--no-enrich": not enrich, + "--instructions": instructions, + "--happened-at": happened_at, + "--attributes": attributes, + "--custom-attributes": custom_attributes, + "--category": category, + "--forceful-relation": forceful_relation, + "--acl": acl, + } + used = [flag for flag, value in unified_only.items() if value] + if used: + print_error(f"{', '.join(used)} appl{'ies' if len(used) == 1 else 'y'} to unified databases only.") if files: # Files are always knowledge sources. Reject every option that would be # silently ignored rather than storing the file the wrong way. Only @@ -156,7 +274,7 @@ def ingest( print_error("--markdown does not apply to file ingest; pass files only.") if not infer: print_error("--infer/--no-infer does not apply to file ingest; pass files only.") - _impl.do_ingest_knowledge_files(files, upsert=upsert, tenant_id=tid, sub_tenant_id=stid) + _impl.do_ingest_knowledge_files(files, upsert=upsert, tenant_id=tid, sub_tenant_id=stid, layout=layout) return if kind == "knowledge": _impl.do_ingest_knowledge_text( @@ -165,6 +283,7 @@ def ingest( source_id=source_id, tenant_id=tid, sub_tenant_id=stid, + layout=layout, ) return _impl.do_ingest_memory( @@ -177,11 +296,14 @@ def ingest( upsert=upsert, tenant_id=tid, sub_tenant_id=stid, + layout=layout, ) def list_items( - kind: str | None = typer.Option(None, "--kind", help="Filter by kind: 'memory' or 'knowledge'."), + kind: str | None = typer.Option( + None, "--kind", help="Filter by kind on a split database: 'memory' or 'knowledge'. Not used on a unified one." + ), page: int | None = typer.Option(None, "--page", help="Page number (1-indexed)."), page_size: int | None = typer.Option(None, "--page-size", help="Items per page (1-100)."), acl: list[str] | None = typer.Option( @@ -221,7 +343,11 @@ def inspect( def delete( ids: list[str] = typer.Argument(help="One or more IDs to delete."), - kind: str = typer.Option("knowledge", "--kind", help="Kind to delete: 'memory' or 'knowledge'."), + kind: str | None = typer.Option( + None, + "--kind", + help="Kind to delete on a split database: 'knowledge' (default) or 'memory'. Not used on a unified one.", + ), database: str | None = typer.Option(None, "--database", "-d", help="Database. Uses default if not specified."), collection: str | None = typer.Option(None, "--collection", help="Collection."), tenant_id: str | None = typer.Option(None, "--tenant-id", hidden=True), @@ -240,7 +366,9 @@ def delete( def relations( source_id: str = typer.Argument(help="Source ID to fetch graph relations for."), - kind: str | None = typer.Option(None, "--kind", help="Corpus: 'memory' or 'knowledge'."), + kind: str | None = typer.Option( + None, "--kind", help="Corpus on a split database: 'memory' or 'knowledge'. Not used on a unified one." + ), limit: int | None = typer.Option(None, "--limit", help="Maximum number of relations to return."), acl: list[str] | None = typer.Option( None, @@ -261,7 +389,9 @@ def relations( def subgraph( source_id: str = typer.Argument(help="Item ID to start from (from 'hydradb query' or 'hydradb list')."), - kind: str | None = typer.Option(None, "--kind", help="Corpus: 'knowledge' (default) or 'memory'."), + kind: str | None = typer.Option( + None, "--kind", help="Corpus on a split database: 'knowledge' (default) or 'memory'. Not used on a unified one." + ), depth: int | None = typer.Option(None, "--depth", help="Hops to traverse (1–10; server default 5)."), max_sources: int | None = typer.Option(None, "--max-sources", help="Cap on members returned (server default 200)."), acl: list[str] | None = typer.Option( @@ -354,9 +484,14 @@ def doctor() -> None: @database_app.command("create") def database_create( database: str = typer.Argument(help="Unique database identifier."), + layout: str | None = typer.Option( + None, + "--type", + help="Storage layout: 'split' (default; separate knowledge and memory corpora selected by --kind) or 'unified' (one corpus; no --kind on later commands).", + ), ) -> None: """Create a new database.""" - _impl.do_database_create(database) + _impl.do_database_create(database, layout) @database_app.command("delete") diff --git a/src/hydradb_cli/hydra/client.py b/src/hydradb_cli/hydra/client.py index e0e346d..28e12b2 100644 --- a/src/hydradb_cli/hydra/client.py +++ b/src/hydradb_cli/hydra/client.py @@ -111,6 +111,17 @@ def _bool_str(value: bool | None) -> str | None: return "true" if value else "false" +#: Storage layouts (PRO-1618). ``split`` is every database that predates the +#: change: a knowledge and a memory corpus, selected by ``type`` on every call. +#: ``unified`` is one corpus; ``type`` is never sent to it. +LAYOUT_SPLIT = "split" +LAYOUT_UNIFIED = "unified" + +#: Item cap on a unified ``POST /context/ingest``, checked locally so an +#: oversized batch is refused before it costs a round trip. +UNIFIED_INGEST_MAX_ITEMS = 100 + + class _Resource: """Base for the ``databases``/``context`` sub-resources.""" @@ -145,7 +156,27 @@ def create( embeddings_dimension: int | None = None, is_embeddings_tenant: bool | None = None, database_metadata_schema: Any | None = None, + layout: str | None = None, ) -> dict: + """Create a database. + + ``layout`` is the storage layout (PRO-1618), sent as the wire field + ``type``: ``split`` (the default, and what every pre-existing database + is) or ``unified`` (one corpus; ``type`` is never sent on later calls). + A layout goes over the raw v2 path so the request does not depend on + which pinned SDK build knows the value; without one this is the + unchanged SDK call. + """ + if layout is not None: + if layout not in (LAYOUT_SPLIT, LAYOUT_UNIFIED): + raise ValueError(f"layout must be '{LAYOUT_SPLIT}' or '{LAYOUT_UNIFIED}', got {layout!r}") + body: dict[str, Any] = {"database": database, "type": layout} + if embeddings_dimension is not None: + body["embeddings_dimension"] = embeddings_dimension + if database_metadata_schema is not None: + body["database_metadata_schema"] = database_metadata_schema + result = self._w._raw_post("/databases", json_body=body) + return result if isinstance(result, dict) else {} resp = self._invoke( self._w._sdk.databases.create, database=database, @@ -163,6 +194,39 @@ def list(self) -> dict: resp = self._invoke(self._w._sdk.databases.list) return _unwrap(resp) + def layouts(self) -> dict[str, str]: + """Every database this key can see, mapped to its storage layout. + + Read from ``GET /databases`` ``details[].type`` (PRO-1618) and memoised + on the wrapper: a layout is fixed at creation, so it cannot go stale + within one process. A database missing from ``details[]`` (an older + server, or one that does not expose the field) is split, which is what + every pre-PRO-1618 database is. + """ + if self._w._layouts is not None: + return self._w._layouts + listed = self.list() + layouts: dict[str, str] = {} + rows = listed.get("details") if isinstance(listed, dict) else None + for row in rows or []: + if isinstance(row, dict) and row.get("database"): + layouts[str(row["database"])] = LAYOUT_UNIFIED if row.get("type") == LAYOUT_UNIFIED else LAYOUT_SPLIT + self._w._layouts = layouts + return layouts + + def layout(self, database: str) -> str: + """The storage layout of one database: ``unified`` or ``split``. + + Split is what a SUCCESSFUL probe reports: a database missing from + ``details[]`` (an older server, or one that does not expose the + field) is what every pre-PRO-1618 database is. A failed probe + propagates instead: answering split on a network, auth or parse + failure would send the split request shape to a database that may + be unified. Failures are not memoised either: ``layouts()`` only + caches a response it actually got, so the next call asks again. + """ + return self.layouts().get(database, LAYOUT_SPLIT) + def collections(self, *, database: str | None = None) -> dict: resp = self._invoke(self._w._sdk.databases.collections, database=self._w._require_database(database)) return _unwrap(resp) @@ -304,6 +368,63 @@ def query( data["request_id"] = request_id return data + def query_unified( + self, + *, + query: str, + operator: str | None = None, + max_results: int | None = None, + mode: str | None = None, + alpha: float | None = None, + recency_bias: float | None = None, + graph_context: bool | None = None, + additional_context: str | None = None, + query_by: str | None = None, + titles: list[str] | None = None, + acl: list[str] | None = None, + follow_forceful_relations: bool | None = None, + database: str | None = None, + collection: str | None = None, + ) -> tuple[dict, str | None]: + """``POST /query`` against a UNIFIED database (PRO-1618). + + Never sends ``type``: a unified database has one corpus, and the server + refuses ``knowledge``/``memory`` there. It goes over the raw v2 path + rather than the SDK because the pinned SDK spells the forceful-relations + switch by its deprecated alias and cannot be relied on to omit ``type``. + + Returns ``(body, request_id)``. ``body`` is the four-key response + (``chunks``, ``graph``, ``forceful_relations``, ``llm_prompt``) exactly + as the server sent it, with nothing added, so ``--output json`` prints + it verbatim; ``request_id`` is lifted from the envelope's ``meta`` for + ``hydradb feedback``, which is the one thing the body cannot carry. + ``request_id`` is the ONLY key read from that ``meta``: a unified + response's meta has no ``tenant_id``, ``sub_tenant_id`` or + ``source_type``. + """ + body = { + key: value + for key, value in { + "database": self._w._require_database(database), + "collection": self._w._resolve_collection(collection), + "query": query, + "operator": operator, + "max_results": max_results, + "mode": mode, + "alpha": alpha, + "recency_bias": recency_bias, + "graph_context": graph_context, + "additional_context": additional_context, + "query_by": query_by, + "titles": titles, + "acl": acl, + "follow_forceful_relations": follow_forceful_relations, + }.items() + if value is not None + } + data, meta = self._w._raw_post_with_meta("/query", json_body=body) + return (data if isinstance(data, dict) else {}), _request_id_of({"meta": meta}) + def ingest( self, *, @@ -371,6 +492,48 @@ def ingest( ) return _unwrap(resp) + def ingest_context( + self, + items: list[dict[str, Any]], + *, + enrich: bool | None = None, + upsert: bool | None = None, + instructions: str | None = None, + database: str | None = None, + collection: str | None = None, + ) -> dict: + """``POST /context/ingest`` on a UNIFIED database (PRO-1618). + + A JSON body whose ``context`` list holds items of exactly one ``text`` + or one ``conversation`` each, in the contract's field names. No + ``type``, no multipart, none of the split-era fields. Items are sent as + given; ``enrich``/``upsert``/``instructions`` are the request-level + defaults for them and travel only when set. + + Returns the 202 payload: ``results[].source_id`` is the item's + ``context_id`` (server-minted when the item carried none). + """ + if not items: + raise ValueError("ingest_context needs at least one item") + if len(items) > UNIFIED_INGEST_MAX_ITEMS: + raise ValueError(f"at most {UNIFIED_INGEST_MAX_ITEMS} items per request, got {len(items)}") + for index, item in enumerate(items): + if not isinstance(item, dict) or ("text" in item) == ("conversation" in item): + raise ValueError(f"context[{index}] must carry exactly one of 'text' or 'conversation'") + body: dict[str, Any] = {"database": self._w._require_database(database)} + coll = self._w._resolve_collection(collection) + if coll: + body["collection"] = coll + body["context"] = list(items) + if enrich is not None: + body["enrich"] = bool(enrich) + if upsert is not None: + body["upsert"] = bool(upsert) + if instructions is not None: + body["instructions"] = instructions + result = self._w._raw_post("/context/ingest", json_body=body) + return result if isinstance(result, dict) else {} + def ingest_many( self, *, @@ -980,6 +1143,9 @@ def __init__( self._token = token self._base_url = base_url or DEFAULT_BASE_URL self._timeout = timeout + # Storage layouts by database name, filled by the first + # ``databases.layouts()`` probe (PRO-1618). None until then. + self._layouts: dict[str, str] | None = None self.databases = _Databases(self) self.context = _Context(self) self.graph = _Graph(self) @@ -1032,6 +1198,17 @@ def _raw_post(self, path: str, *, json_body: Any) -> Any: ``API-Version: 2`` headers, same shape-based unwrapping, same translated error type, so a caller cannot tell it from an SDK call. """ + return self._raw_post_with_meta(path, json_body=json_body)[0] + + def _raw_post_with_meta(self, path: str, *, json_body: Any) -> tuple[Any, dict]: + """:meth:`_raw_post`, also returning the envelope's ``meta``. + + ``_unwrap_payload`` keeps ``data`` and drops ``meta``, which is where + ``request_id`` lives. A unified ``/query`` must hand back ``data`` + untouched (the four-key body, printed verbatim) AND surface the request + id for ``hydradb feedback``, so this variant returns both. ``meta`` is + ``{}`` when the response was not an envelope. + """ headers = { "Authorization": f"Bearer {self._token}", "API-Version": "2", @@ -1054,7 +1231,8 @@ def _raw_post(self, path: str, *, json_body: Any) -> Any: if response.is_error: raise HydraDBClientError(response.status_code, _stringify_body(body)) - return _unwrap_payload(body) + meta = body.get("meta") if isinstance(body, dict) else None + return _unwrap_payload(body), (meta if isinstance(meta, dict) else {}) def _require_database(self, database: str | None) -> str: db = database or self.default_database diff --git a/tests/golden/query_unified.json b/tests/golden/query_unified.json new file mode 100644 index 0000000..618d6c0 --- /dev/null +++ b/tests/golden/query_unified.json @@ -0,0 +1,99 @@ +{ + "success": true, + "data": { + "chunks": [ + { + "chunk_id": "ck_policy_3", + "context_id": "refund-policy", + "score": 0.91, + "content": "Refunds are processed within 30 days of purchase by the Finance Department.", + "enrichment": "Refund window is 30 days; Finance owns refund processing.", + "enrichment_kind": "business_knowledge", + "temporal": [ + { + "content": "Refund policy effective_from June 2026. Start: 2026-06-01", + "start_date": "2026-06-01", + "end_date": null + } + ] + }, + { + "chunk_id": "ck_chat_1", + "context_id": "chat-2026-07-29", + "score": 0.84, + "content": "user: Keep refund answers short please\nassistant: Got it.", + "enrichment": "User prefers short answers about refunds.", + "enrichment_kind": "user_preference" + } + ], + "graph": [ + { + "origin": "query_path", + "triplets": [ + { + "source": { + "entity_id": "ent_refunds", + "name": "Refund Processing" + }, + "relation": { + "predicate": "managed by", + "context": "Refund processing is managed by the Finance Department.", + "relationship_id": "rel_managed_by", + "chunk_id": "ck_policy_3" + }, + "target": { + "entity_id": "ent_finance", + "name": "Finance Department" + } + } + ], + "path_summary": "Refund processing is managed by the Finance Department." + }, + { + "origin": "chunk_relation", + "triplets": [ + { + "source": { + "entity_id": "ent_user", + "name": "User" + }, + "relation": { + "predicate": "prefers", + "context": "The user prefers short answers about refunds.", + "relationship_id": "rel_prefers", + "chunk_id": "ck_chat_1" + }, + "target": { + "entity_id": "ent_short", + "name": "short answers" + } + } + ], + "path_summary": "The user prefers short answers about refunds." + } + ], + "forceful_relations": [ + { + "via": { + "from": "refund-policy", + "to": "refund-faq" + }, + "chunk": { + "chunk_id": "ck_faq_1", + "context_id": "refund-faq", + "score": 0, + "content": "FAQ: refunds to a card take 5 to 7 business days to appear." + } + } + ], + "llm_prompt": "# Query results\n\n**Query:** who owns refund processing?\n**Found:** 2 results across 2 sources · 2 related facts · 1 temporal fact · 1 forceful relation\nCite a result by its number in brackets, e.g. [1].\n\n## Results\n\n### 1. Refund policy\n- **Relevance:** 0.91 · **Collection:** support · **Type:** file · **Category:** business_knowledge\n- **Id:** refund-policy · **Last updated:** 2026-07-02\n\nRefunds are processed within 30 days of purchase by the Finance Department.\n\n**Enrichment:** Refund window is 30 days; Finance owns refund processing.\n\n---\n\n### 2. Support chat with Priya\n- **Relevance:** 0.84 · **Collection:** support · **Type:** message · **Category:** user_preference\n- **Id:** chat-2026-07-29 · **Last updated:** 2026-07-29\n\nuser: Keep refund answers short please\nassistant: Got it.\n\n**Enrichment:** User prefers short answers about refunds.\n\n## Forceful relations\n\nLinked to a result by the author at ingest time (forceful_relations), not by relevance to this query.\n\n### R1. Refund FAQ\n- **Linked from:** refund-policy · **Collection:** support\n- **Id:** refund-faq\n\nFAQ: refunds to a card take 5 to 7 business days to appear.\n\n## Related facts\n\n- [P1] **Refund Processing** -managed by→ **Finance Department** (query path, relevance 0.81) [1]\n Refund processing is managed by the Finance Department.\n- [P2] **User** -prefers→ **short answers** (chunk relation, relevance 0.74) [2]\n The user prefers short answers about refunds.\n\n## Temporal facts\n\n- **Refund policy** *effective_from* → **June 2026** (from 2026-06-01, precision: month, status: ongoing; evidence: \"from June\") [1]\n\n## Sources\n\n1. **Refund policy** (file, id: refund-policy) · https://docs.acme.com/refunds · updated 2026-07-02\n2. **Support chat with Priya** (message, id: chat-2026-07-29) · updated 2026-07-29\n3. **Refund FAQ** (id: refund-faq)" + }, + "error": null, + "meta": { + "request_id": "ab5a04df-d3c4-419c-9669-d1ea2c3f9c51", + "api_version": "2.0.1", + "latency_ms": null, + "database": "acme_corp", + "collection": "support" + } +} diff --git a/tests/test_unified.py b/tests/test_unified.py new file mode 100644 index 0000000..cc00c8d --- /dev/null +++ b/tests/test_unified.py @@ -0,0 +1,1268 @@ +"""Unified databases (PRO-1618). + +A database is ``split`` (knowledge + memory corpora, ``type`` on every call) +or ``unified`` (one corpus; ``type`` is never sent). The CLI reads the layout +once per database from ``GET /databases`` ``details[].type`` and branches on +THAT, never on a request flag. Everything here is about the unified side, and +about the split side staying exactly as it was. +""" + +import io +import json +import re +from pathlib import Path +from unittest.mock import MagicMock, patch + +import httpx +import pytest +import typer +from hydra_db import HydraDB as _SdkHydraDB +from rich.console import Console +from typer.testing import CliRunner + +import hydradb_cli.config +import hydradb_cli.output +from hydradb_cli.commands import _impl +from hydradb_cli.config import save_config +from hydradb_cli.hydra import HydraDB, HydraDBClientError +from hydradb_cli.main import app + +runner = CliRunner() +GOLDEN = Path(__file__).parent / "golden" +# A real unified /query envelope, rendered by the server's own handler test +# (hydradb-application PRO-1618). Its ``meta.request_id`` is not used: each +# test hands the CLI its own request_id through the wrapper or the envelope. +UNIFIED_ENVELOPE = json.loads((GOLDEN / "query_unified.json").read_text()) +UNIFIED_BODY = UNIFIED_ENVELOPE["data"] +SPLIT_BODY = json.loads((GOLDEN / "query.json").read_text()) + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") +_WIDE = {"COLUMNS": "200", "TERM": "dumb", "NO_COLOR": "1"} +_HYDRA_ENV_VARS = ( + "HYDRADB_API_KEY", + "HYDRADB_DATABASE", + "HYDRADB_COLLECTION", + "HYDRADB_BASE_URL", + "HYDRADB_OUTPUT", + "HYDRADB_TENANT_ID", + "HYDRADB_SUB_TENANT_ID", + "HYDRADB_API_URL", + "HYDRA_DB_API_KEY", + "HYDRA_DB_TENANT_ID", + "HYDRA_DB_SUB_TENANT_ID", + "HYDRA_DB_BASE_URL", + "HYDRA_OPENCLAW_API_KEY", + "HYDRA_OPENCLAW_TENANT_ID", +) + +# One ``context[]`` item carrying every documented field, in the contract's +# exact names. The CLI test and the wrapper test both pin against it. +FULL_ITEM = { + "context_id": "policy-1", + "title": "Refund policy", + "text": "Refund policy: 30-day window.", + "enrich": True, + "upsert": True, + "instructions": "keep the window", + "happened_at": "2026-07-29", + "attributes": {"team": "support"}, + "custom_attributes": {"source_app": "wiki"}, + "context_category": "business_knowledge", + "forceful_relations": {"ids": ["chat-w1"]}, + "acl": ["user_email:a@x.com", "domain:acme.com"], +} + +INGEST_202 = { + "success": True, + "message": "queued", + "results": [ + { + "source_id": "policy-1", + "title": "Refund policy", + "status": "queued", + "infer": True, + "error": None, + "error_code": None, + } + ], + "success_count": 1, + "failed_count": 0, +} + +EMPTY_BODY = {"chunks": [], "graph": [], "forceful_relations": [], "llm_prompt": ""} + + +@pytest.fixture(autouse=True) +def clean_config(tmp_path, monkeypatch): + config_dir = tmp_path / ".hydradb" + monkeypatch.setattr("hydradb_cli.config.CONFIG_DIR", config_dir) + monkeypatch.setattr("hydradb_cli.config.CONFIG_FILE", config_dir / "config.json") + for var in _HYDRA_ENV_VARS: + monkeypatch.delenv(var, raising=False) + hydradb_cli.output._warned_deprecations.clear() + hydradb_cli.config._warned_env_aliases.clear() + yield + + +def _auth(): + save_config(api_key="test-key-abcdef1234567890", tenant_id="t1") + + +def _mock(layout: str, **returns): + """A MagicMock wrapper whose layout probe answers ``layout``.""" + w = MagicMock() + w.databases.layout.return_value = layout + for dotted, value in returns.items(): + resource, method = dotted.split(".") + getattr(getattr(w, resource), method).return_value = value + return w + + +def _patch(w): + return patch("hydradb_cli.commands._impl.get_wrapper", return_value=w) + + +def _plain(text: str) -> str: + return re.sub(r"\s+", " ", _ANSI_RE.sub("", text)) + + +def _lines(result) -> list[str]: + return [_ANSI_RE.sub("", line) for line in result.output.splitlines()] + + +def _real_wrapper(sdk_handler, database="db_test", collection="col_test") -> HydraDB: + """The real wrapper: the SDK on a mock transport, the raw path untouched.""" + w = HydraDB(token="x", base_url="http://test.local", database=database, collection=collection) + w._sdk = _SdkHydraDB( + token="x", + base_url="http://test.local", + httpx_client=httpx.Client(transport=httpx.MockTransport(sdk_handler)), + ) + return w + + +def _databases_envelope(details: list[dict]) -> dict: + return { + "success": True, + "meta": {}, + "data": {"databases": [d["database"] for d in details], "details": details}, + } + + +def _sdk_500(request): + return httpx.Response(500, json={"success": False, "error": {"message": "the SDK path must not be used"}}) + + +class _Capture: + """Stands in for ``httpx.post`` on the raw path and records every call.""" + + def __init__(self, status: int = 200, body: dict | None = None): + self.status = status + self.body = body if body is not None else {"success": True, "data": {}, "meta": {}} + self.calls: list[dict] = [] + + def __call__(self, url, *, headers=None, json=None, timeout=None, **_ignored): + self.calls.append({"url": url, "headers": headers, "json": json}) + return httpx.Response(self.status, json=self.body, request=httpx.Request("POST", url)) + + +def _capture_post(monkeypatch, status: int = 200, body: dict | None = None) -> _Capture: + capture = _Capture(status, body) + monkeypatch.setattr("hydradb_cli.hydra.client.httpx.post", capture) + return capture + + +class _WatchedMeta(dict): + """An envelope ``meta`` that records every key read from it (``*`` for a + read of the whole mapping), so a test can pin what the unified path reads.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.read: list[str] = [] + + def __getitem__(self, key): + self.read.append(key) + return super().__getitem__(key) + + def get(self, key, default=None): + self.read.append(key) + return super().get(key, default) + + def __contains__(self, key): + self.read.append(key) + return super().__contains__(key) + + def __iter__(self): + self.read.append("*") + return super().__iter__() + + def keys(self): + self.read.append("*") + return super().keys() + + def items(self): + self.read.append("*") + return super().items() + + def values(self): + self.read.append("*") + return super().values() + + +def _path(origin: str | None, chunk_id: str, source: str, target: str, summary: str) -> dict: + """One ``graph[]`` path of a single hop extracted from ``chunk_id``.""" + path: dict = { + "triplets": [ + { + "source": {"entity_id": f"ent_{source}", "name": source}, + "relation": { + "predicate": "links", + "context": f"{source} links {target}.", + "relationship_id": f"rel_{source}", + "chunk_id": chunk_id, + }, + "target": {"entity_id": f"ent_{target}", "name": target}, + } + ], + "path_summary": summary, + } + if origin is not None: + path["origin"] = origin + return path + + +# ── the layout probe ───────────────────────────────────────────────────────── + + +class TestLayoutProbe: + def test_layout_reads_details_and_memoises(self): + seen = [] + + def handler(request): + seen.append((request.method, request.url.path)) + return httpx.Response( + 200, + json=_databases_envelope([{"database": "a", "type": "unified"}, {"database": "b", "type": "split"}]), + ) + + w = _real_wrapper(handler) + assert w.databases.layout("a") == "unified" + assert w.databases.layout("b") == "split" + assert w.databases.layout("missing") == "split" + assert seen == [("GET", "/databases")], "one probe per wrapper" + + def test_no_details_means_every_database_is_split(self): + w = _real_wrapper( + lambda r: httpx.Response(200, json={"success": True, "meta": {}, "data": {"databases": ["a"]}}) + ) + assert w.databases.layout("a") == "split" + + def test_a_failed_probe_raises_and_is_not_memoised(self): + failing = {"on": True} + + def handler(request): + if failing["on"]: + return httpx.Response(401, json={"success": False, "error": {"message": "bad key"}}) + return httpx.Response(200, json=_databases_envelope([{"database": "a", "type": "unified"}])) + + w = _real_wrapper(handler) + # A failed probe is an error, not a guess: answering split here would + # send the split request shape to a database that may be unified. + with pytest.raises(HydraDBClientError): + w.databases.layout("a") + failing["on"] = False + assert w.databases.layout("a") == "unified", "a failed probe must not be memoised" + + def test_a_failed_probe_surfaces_as_a_cli_error(self): + _auth() + w = _mock("split") + w.databases.layout.side_effect = HydraDBClientError(401, "bad key") + with _patch(w): + result = runner.invoke(app, ["list"], env=_WIDE) + assert result.exit_code != 0 + assert "Authentication failed" in result.output + + def test_a_mocked_wrapper_reads_as_split(self): + # Compared by value: a MagicMock layout is not "unified", so every + # existing test that mocks the wrapper keeps exercising the split path. + assert _impl._is_unified(MagicMock(), "anything") is False + + def test_database_layout_resolves_the_scope_first(self): + with _patch(_mock("unified")), pytest.raises(typer.Exit): + _impl.database_layout(None) + _auth() + with _patch(_mock("unified")): + assert _impl.database_layout(None) == ("t1", "unified") + with _patch(_mock("split")): + assert _impl.database_layout("other") == ("other", "split") + + +# ── wrapper: unified query ─────────────────────────────────────────────────── + + +class TestUnifiedQueryWrapper: + def test_posts_no_type_and_returns_the_body_verbatim_with_the_request_id(self, monkeypatch): + capture = _capture_post( + monkeypatch, body={"success": True, "data": UNIFIED_BODY, "meta": {"request_id": "req-1", "latency_ms": 12}} + ) + w = _real_wrapper(_sdk_500) + body, request_id = w.context.query_unified( + query="pro plan", + operator="and", + query_by="text", + max_results=5, + mode="fast", + titles=["Q3"], + acl=["a@x.com"], + follow_forceful_relations=False, + ) + assert body == UNIFIED_BODY, "the four keys, nothing added, nothing dropped" + assert request_id == "req-1" + call = capture.calls[0] + assert call["url"] == "http://test.local/query" + assert call["headers"]["API-Version"] == "2" + assert call["json"] == { + "database": "db_test", + "collection": "col_test", + "query": "pro plan", + "operator": "and", + "max_results": 5, + "mode": "fast", + "query_by": "text", + "titles": ["Q3"], + "acl": ["a@x.com"], + "follow_forceful_relations": False, + } + assert "type" not in call["json"] + + def test_unset_fields_are_omitted_not_sent_as_null(self, monkeypatch): + capture = _capture_post(monkeypatch, body={"success": True, "data": EMPTY_BODY, "meta": {}}) + body, request_id = _real_wrapper(_sdk_500).context.query_unified(query="q") + assert capture.calls[0]["json"] == {"database": "db_test", "collection": "col_test", "query": "q"} + assert body == EMPTY_BODY + assert request_id is None + + def test_a_refusal_is_a_client_error(self, monkeypatch): + _capture_post(monkeypatch, status=400, body={"success": False, "error": {"message": "knowledge is not valid"}}) + with pytest.raises(HydraDBClientError) as excinfo: + _real_wrapper(_sdk_500).context.query_unified(query="q") + assert excinfo.value.status_code == 400 + assert "knowledge is not valid" in str(excinfo.value.detail) + + +# ── wrapper: unified ingest ────────────────────────────────────────────────── + + +class TestUnifiedIngestWrapper: + def test_posts_the_exact_json_body(self, monkeypatch): + capture = _capture_post(monkeypatch, status=202, body={"success": True, "data": INGEST_202, "meta": {}}) + out = _real_wrapper(_sdk_500).context.ingest_context([FULL_ITEM]) + call = capture.calls[0] + assert call["url"] == "http://test.local/context/ingest" + assert call["headers"]["API-Version"] == "2" + assert call["json"] == {"database": "db_test", "collection": "col_test", "context": [FULL_ITEM]} + for forbidden in ("type", "items", "contexts", "memories", "app_knowledge", "documents"): + assert forbidden not in call["json"], forbidden + # The 202: results[].source_id is the context id. + assert out["results"][0]["source_id"] == "policy-1" + assert out["success_count"] == 1 and out["failed_count"] == 0 + + def test_request_level_defaults_travel_only_when_set(self, monkeypatch): + capture = _capture_post(monkeypatch, status=202, body={"success": True, "data": INGEST_202, "meta": {}}) + w = _real_wrapper(_sdk_500, collection=None) + w.context.ingest_context([{"text": "x"}]) + assert capture.calls[0]["json"] == {"database": "db_test", "context": [{"text": "x"}]} + w.context.ingest_context([{"text": "x"}], enrich=False, upsert=False, instructions="be brief") + assert capture.calls[1]["json"] == { + "database": "db_test", + "context": [{"text": "x"}], + "enrich": False, + "upsert": False, + "instructions": "be brief", + } + + def test_a_conversation_item_is_sent_as_given(self, monkeypatch): + capture = _capture_post(monkeypatch, status=202, body={"success": True, "data": INGEST_202, "meta": {}}) + turns = [{"role": "user", "content": "hi", "name": "soham"}, {"role": "assistant", "content": "hello"}] + _real_wrapper(_sdk_500).context.ingest_context([{"context_id": "chat-w1", "conversation": turns}]) + assert capture.calls[0]["json"]["context"] == [{"context_id": "chat-w1", "conversation": turns}] + + def test_refuses_an_item_with_both_or_neither_shape(self): + w = _real_wrapper(_sdk_500) + with pytest.raises(ValueError, match="exactly one"): + w.context.ingest_context([{"text": "a", "conversation": []}]) + with pytest.raises(ValueError, match=r"context\[1\]"): + w.context.ingest_context([{"text": "a"}, {"title": "no body"}]) + with pytest.raises(ValueError, match="at least one"): + w.context.ingest_context([]) + + def test_refuses_more_than_100_items(self): + with pytest.raises(ValueError, match="100"): + _real_wrapper(_sdk_500).context.ingest_context([{"text": "x"}] * 101) + + def test_a_refusal_is_a_client_error(self, monkeypatch): + _capture_post(monkeypatch, status=400, body={"success": False, "error": {"message": "context[0]: too large"}}) + with pytest.raises(HydraDBClientError) as excinfo: + _real_wrapper(_sdk_500).context.ingest_context([{"text": "x"}]) + assert excinfo.value.status_code == 400 + + +# ── hydradb query ──────────────────────────────────────────────────────────── + + +class TestUnifiedQueryCommand: + def test_goes_over_query_unified_and_never_sends_a_kind(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + result = runner.invoke( + app, + [ + "query", + "pro plan", + "--no-follow-forceful-relations", + "--title", + "Q3", + "--operator", + "and", + "-n", + "5", + ], + ) + assert result.exit_code == 0, result.output + w.context.query.assert_not_called() + kwargs = w.context.query_unified.call_args.kwargs + assert "kind" not in kwargs and "type" not in kwargs + assert kwargs["follow_forceful_relations"] is False + assert kwargs["titles"] == ["Q3"] + assert kwargs["operator"] == "and" and kwargs["query_by"] == "text" + assert kwargs["max_results"] == 5 + assert kwargs["database"] == "t1" + + def test_the_follow_switch_is_omitted_unless_given(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + runner.invoke(app, ["query", "pro plan"]) + assert w.context.query_unified.call_args.kwargs["follow_forceful_relations"] is None + with _patch(w): + runner.invoke(app, ["query", "pro plan", "--follow-forceful-relations"]) + assert w.context.query_unified.call_args.kwargs["follow_forceful_relations"] is True + + def test_human_output_renders_chunks_graph_and_forceful_relations(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + result = runner.invoke(app, ["query", "pro plan"], env=_WIDE) + assert result.exit_code == 0, result.output + out = _plain(result.output) + assert "Found 2 result(s)" in out + # chunks[]: context_id, score, content, enrichment (a string) with + # enrichment_kind beside it, temporal facts + assert "refund-policy" in out and "91%" in out + assert "Refunds are processed within 30 days of purchase by the Finance Department." in out + assert "enrichment (business_knowledge): Refund window is 30 days; Finance owns refund processing." in out + assert "chat-2026-07-29" in out and "84%" in out and "Keep refund answers short please" in out + assert "enrichment (user_preference): User prefers short answers about refunds." in out + # a temporal fact with only a start date prints only that side + assert "temporal: Refund policy effective_from June 2026. Start: 2026-06-01 [2026-06-01]" in out + # graph[]: grouped by origin, path_summary + triplets. A query-path hop + # cites the returned chunk it came from; a chunk relation is listed + # under the chunk it hangs under. + assert "/// Graph: 1 query path(s)" in out + assert "Refund processing is managed by the Finance Department." in out + assert "Refund Processing -> managed by -> Finance Department [1]" in out + assert "/// Graph: 1 chunk relation path(s)" in out + assert "[2] chat-2026-07-29" in out + assert "The user prefers short answers about refunds." in out + assert "User -> prefers -> short answers" in out + assert out.index("/// Graph: 1 query path(s)") < out.index("/// Graph: 1 chunk relation path(s)") + # forceful_relations[]: R label, via from/to and the pulled-in chunk + assert "/// Forceful relations: 1 chunk(s)" in out + assert "R1" in out and "refund-faq" in out + assert "refunds to a card take 5 to 7 business days" in out and "0%" in out + assert "Related" not in out + # the markdown prompt is not dumped into the structured view + assert "# Query results" not in out and "**Enrichment:**" not in out + # feedback stays reachable + assert "hydradb feedback req-1" in out + + def test_json_prints_the_body_verbatim(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + result = runner.invoke(app, ["--output", "json", "query", "pro plan"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == UNIFIED_BODY + + def test_llm_prints_the_prompt_verbatim_on_stdout(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + result = runner.invoke(app, ["query", "pro plan", "--llm"]) + assert result.exit_code == 0, result.output + assert result.stdout == UNIFIED_BODY["llm_prompt"] + "\n" + assert "req-1" in result.stderr, "the feedback hint goes to stderr so the prompt can be piped" + + def test_llm_yields_to_json_output(self): + _auth() + w = _mock("unified", **{"context.query_unified": (UNIFIED_BODY, "req-1")}) + with _patch(w): + result = runner.invoke(app, ["--output", "json", "query", "pro plan", "--llm"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == UNIFIED_BODY + + def test_an_empty_result_still_prints_the_request_id(self): + _auth() + w = _mock("unified", **{"context.query_unified": (EMPTY_BODY, "req-1")}) + with _patch(w): + result = runner.invoke(app, ["query", "nothing"]) + assert result.exit_code == 0, result.output + assert "No relevant results found." in result.output + assert "req-1" in result.output + + def test_kind_is_refused_on_a_unified_database(self): + _auth() + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["query", "x", "--kind", "memory"]) + assert result.exit_code != 0 + assert "unified" in result.output and "--kind" in result.output + w.context.query.assert_not_called() + w.context.query_unified.assert_not_called() + + def test_deprecated_recall_aliases_are_refused_on_a_unified_database(self): + _auth() + for argv in (["recall", "full", "x"], ["recall", "preferences", "x"]): + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, argv) + assert result.exit_code != 0, argv + assert "unified" in result.output + w.context.query.assert_not_called() + + def test_llm_and_follow_flags_are_refused_on_a_split_database(self): + _auth() + w = _mock("split", **{"context.query": {"chunks": []}}) + for extra in (["--llm"], ["--no-follow-forceful-relations"], ["--follow-forceful-relations"]): + with _patch(w): + result = runner.invoke(app, ["query", "x", *extra]) + assert result.exit_code != 0, extra + assert "unified databases only" in result.output + w.context.query.assert_not_called() + + def test_split_query_call_is_unchanged(self): + _auth() + w = _mock( + "split", **{"context.query": {"chunks": [{"chunk_content": "Pricing is $29", "relevancy_score": 0.9}]}} + ) + with _patch(w): + result = runner.invoke(app, ["query", "pricing", "--kind", "knowledge"]) + assert result.exit_code == 0, result.output + assert "Pricing" in result.output + kwargs = w.context.query.call_args.kwargs + assert kwargs["kind"] == "knowledge" + assert "follow_forceful_relations" not in kwargs and "llm" not in kwargs + w.context.query_unified.assert_not_called() + + def test_only_the_request_id_is_read_from_the_unified_meta(self, monkeypatch): + """A unified /query ``meta`` has no ``tenant_id``, ``sub_tenant_id`` or + ``source_type`` (PRO-1618): nothing on the unified path may read them, + and ``request_id`` is the one key it needs, in every output mode.""" + monkeypatch.setenv("HYDRADB_API_KEY", "x") + monkeypatch.setenv("HYDRADB_DATABASE", "db_test") + meta = _WatchedMeta( + {"request_id": "req-7", "api_version": "2", "latency_ms": 9, "database": "db_test", "collection": "c"} + ) + wrapper = _real_wrapper( + lambda r: httpx.Response(200, json=_databases_envelope([{"database": "db_test", "type": "unified"}])) + ) + monkeypatch.setattr(wrapper, "_raw_post_with_meta", lambda path, *, json_body: (UNIFIED_BODY, meta)) + outputs = {} + for mode, argv in ( + ("human", ["query", "pro plan"]), + ("llm", ["query", "pro plan", "--llm"]), + ("json", ["--output", "json", "query", "pro plan"]), + ): + with patch("hydradb_cli.commands._impl.get_wrapper", return_value=wrapper): + result = runner.invoke(app, argv, env=_WIDE) + assert result.exit_code == 0, (mode, result.output) + outputs[mode] = result + assert set(meta.read) == {"request_id"}, meta.read + assert "hydradb feedback req-7" in outputs["human"].output + assert "req-7" in outputs["llm"].stderr + assert json.loads(outputs["json"].stdout) == UNIFIED_BODY + + def test_end_to_end_json_is_the_server_body_verbatim(self, monkeypatch): + """Real wrapper: the probe over the SDK transport, the query over the raw path.""" + monkeypatch.setenv("HYDRADB_API_KEY", "x") + monkeypatch.setenv("HYDRADB_DATABASE", "db_test") + monkeypatch.setenv("HYDRADB_COLLECTION", "col_test") + probes = [] + + def sdk_handler(request): + probes.append(request.url.path) + return httpx.Response(200, json=_databases_envelope([{"database": "db_test", "type": "unified"}])) + + capture = _capture_post( + monkeypatch, body={"success": True, "data": UNIFIED_BODY, "meta": {"request_id": "req-9"}} + ) + with patch("hydradb_cli.commands._impl.get_wrapper", return_value=_real_wrapper(sdk_handler)): + result = runner.invoke(app, ["--output", "json", "query", "pro plan"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == UNIFIED_BODY + assert probes == ["/databases"] + assert capture.calls[0]["url"] == "http://test.local/query" + assert "type" not in capture.calls[0]["json"] + + +# ── the renderer ───────────────────────────────────────────────────────────── + + +class TestQueryRenderer: + @staticmethod + def _render(renderable, width: int = 120) -> str: + buffer = io.StringIO() + Console(file=buffer, width=width, force_terminal=False, no_color=True).print(renderable) + return buffer.getvalue() + + def test_shape_detection_is_by_body_not_by_layout(self): + assert _impl._is_unified_query_body(UNIFIED_BODY) + assert _impl._is_unified_query_body(EMPTY_BODY) + assert not _impl._is_unified_query_body(SPLIT_BODY) + assert not _impl._is_unified_query_body({"chunks": []}) + assert not _impl._is_unified_query_body({"chunks": [], "graph_context": {"query_paths": []}}) + + def test_a_split_body_carrying_graph_and_forceful_relations_objects_is_split(self): + # A split body carries `graph` ({paths}) and `forceful_relations` + # ({declared, inferred}) beside graph_context: the unified keys, as + # objects. Shape detection goes by type, never by the key being there. + split = { + **SPLIT_BODY, + "graph_context": {"query_paths": [], "chunk_relations": []}, + "graph": {"paths": []}, + "forceful_relations": {"declared": [], "inferred": []}, + } + assert not _impl._is_unified_query_body(split) + assert _impl._is_unified_query_body({"chunks": [], "forceful_relations": []}) + out = self._render(_impl._format_query_result(split)) + assert "Pricing is $29/mo" in out + assert "/// Forceful relations" not in out and "/// Graph" not in out + + def test_unified_content_enrichment_and_forceful_rows_are_never_trimmed(self): + # The unified answer is not compacted anywhere: long content, + # enrichment and a forceful chunk's content all reach the screen whole. + long = " ".join(["word"] * 150) + chunk = {"chunk_id": "c1", "context_id": "doc-1", "score": 0.5} + body = { + "chunks": [{**chunk, "content": long + " CONTENTTAIL", "enrichment": long + " ENRICHTAIL"}], + "graph": [], + "forceful_relations": [ + {"via": {"from": "doc-1", "to": "doc-2"}, "chunk": {**chunk, "content": long + " FORCEFULTAIL"}} + ], + "llm_prompt": "", + } + out = self._render(_impl._format_query_result(body)) + for tail in ("CONTENTTAIL", "ENRICHTAIL", "FORCEFULTAIL"): + assert tail in out, tail + assert "..." not in out + + def test_the_split_golden_renders_through_the_split_renderer_unchanged(self): + out = self._render(_impl._format_query_result(SPLIT_BODY)) + assert "Found 1 result(s)" in out + assert "92%" in out and "Pricing Doc" in out and "Pricing is $29/mo" in out + for marker in ("enrichment", "/// Graph", "/// Forceful relations", "context_id"): + assert marker not in out, marker + + def test_a_unified_body_through_the_shared_entry_point_renders_unified(self): + # A server that does not list a database's layout still answers a + # type-less query on a unified one in the unified shape, and the + # renderer must still read it. + out = self._render(_impl._format_query_result({**UNIFIED_BODY, "request_id": "req-2"}), width=200) + assert "Refund Processing -> managed by -> Finance Department" in out + assert "enrichment (user_preference)" in out + assert "req-2" in out + + def test_the_forceful_relations_section_is_hidden_when_empty(self): + out = self._render(_impl._format_unified_query_result({**UNIFIED_BODY, "forceful_relations": []})) + assert "Found 2 result(s)" in out and "/// Graph" in out + assert "Forceful relations" not in out and "refund-faq" not in out + + def test_the_old_relations_key_is_not_read(self): + # `relations` was renamed `forceful_relations`; there is no fallback. + old = {key: value for key, value in UNIFIED_BODY.items() if key != "forceful_relations"} + old["relations"] = UNIFIED_BODY["forceful_relations"] + out = self._render(_impl._format_unified_query_result(old)) + assert "Found 2 result(s)" in out + assert "Forceful relations" not in out and "business days" not in out + only_old = {"chunks": [], "graph": [], "relations": UNIFIED_BODY["forceful_relations"], "llm_prompt": ""} + assert "No relevant results found." in self._render(_impl._format_unified_query_result(only_old)) + + def test_graph_is_grouped_by_origin_with_positional_labels(self): + body = { + **UNIFIED_BODY, + "graph": [ + _path("chunk_relation", "ck_faq_1", "Comment", "Fix", "The comment names the fix."), + _path("query_path", "ck_not_returned", "Alpha", "Beta", "Alpha links Beta."), + _path(None, "ck_policy_3", "Gamma", "Delta", "Gamma links Delta."), + ], + } + out = _plain(self._render(_impl._format_unified_query_result(body), width=200)) + # One group per origin, query paths first. P labels are positions in + # graph[], which is how llm_prompt numbers paths. + assert "/// Graph: 1 query path(s)" in out + assert re.search(r"P2\W+Alpha links Beta\.", out) + # a hop whose chunk is not in the result cites nothing + assert "Alpha -> links -> Beta" in out and "Alpha -> links -> Beta [" not in out + # a chunk relation hangs under the chunk its hops came from, here a + # forceful-relation chunk, by its R label + assert "/// Graph: 1 chunk relation path(s)" in out + assert re.search(r"P1\W+\[R1\] refund-faq\W+The comment names the fix\.", out) + # a path with no origin is still shown, in its own group, not guessed into one + assert "/// Graph: 1 path(s) with no known origin" in out + assert re.search(r"P3\W+Gamma links Delta\.", out) and "Gamma -> links -> Delta [1]" in out + assert out.index("query path(s)") < out.index("chunk relation path(s)") < out.index("no known origin") + + def test_enrichment_is_a_string_with_its_kind_beside_it(self): + chunk = {"context_id": "c1", "score": 0.5, "content": "x"} + cases = ( + # enrichment_kind is present even when there is no enrichment + ({"enrichment_kind": "decision_trace"}, "enrichment (decision_trace)", "enrichment (decision_trace):"), + ({"enrichment": "Owns refunds."}, "enrichment: Owns refunds.", "enrichment ("), + ( + {"enrichment": "Owns refunds.", "enrichment_kind": "business_knowledge"}, + "enrichment (business_knowledge): Owns refunds.", + None, + ), + ) + for fields, present, absent in cases: + body = {**EMPTY_BODY, "chunks": [{**chunk, **fields}]} + out = self._render(_impl._format_unified_query_result(body)) + assert present in out, fields + if absent: + assert absent not in out, fields + + def test_no_enrichment_line_without_enrichment_or_kind(self): + body = {**EMPTY_BODY, "chunks": [{"context_id": "c1", "score": 0.5, "content": "x"}]} + assert "enrichment" not in self._render(_impl._format_unified_query_result(body)) + + def test_the_old_enrichment_object_is_not_read(self): + # `enrichment` was `{text, kind}`; it is a string now, with the kind in + # `enrichment_kind`. An object is not rendered, and does not crash. + body = { + **EMPTY_BODY, + "chunks": [ + {"context_id": "c1", "content": "x", "enrichment": {"text": "stale text", "kind": "decision_trace"}} + ], + } + out = self._render(_impl._format_unified_query_result(body)) + assert "Found 1 result(s)" in out + assert "enrichment" not in out and "stale text" not in out + + +# ── hydradb ingest ─────────────────────────────────────────────────────────── + + +class TestUnifiedIngestCommand: + def test_text_posts_one_context_item_with_every_field(self): + _auth() + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke( + app, + [ + "ingest", + "--text", + "Refund policy: 30-day window.", + "--context-id", + "policy-1", + "--title", + "Refund policy", + "--instructions", + "keep the window", + "--happened-at", + "2026-07-29", + "--attributes", + '{"team": "support"}', + "--custom-attributes", + '{"source_app": "wiki"}', + "--category", + "business_knowledge", + "--forceful-relation", + "chat-w1", + "--acl", + "user_email:a@x.com", + "--acl", + "domain:acme.com", + ], + ) + assert result.exit_code == 0, result.output + w.context.ingest.assert_not_called() + w.context.ingest_many.assert_not_called() + args, kwargs = w.context.ingest_context.call_args + assert args == ([FULL_ITEM],) + assert kwargs == {"database": "t1", "collection": None} + out = _plain(result.output) + assert "Context queued (1 success, 0 failed)" in out + assert "Context ID: policy-1 (queued)" in out + + def test_conversation_file(self, tmp_path): + _auth() + turns = [ + {"role": "user", "content": "Keep answers short please", "name": "soham"}, + {"role": "assistant", "content": "Got it."}, + {"role": "system", "content": "Never store account numbers"}, + ] + f = tmp_path / "conv.json" + f.write_text(json.dumps(turns)) + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--conversation-file", str(f), "--context-id", "chat-w1"]) + assert result.exit_code == 0, result.output + item = w.context.ingest_context.call_args.args[0][0] + assert item == {"context_id": "chat-w1", "conversation": turns, "enrich": True, "upsert": True} + assert "conversation, 3 turn(s)" in _plain(result.output) + + def test_defaults_are_explicit_and_nothing_else_is_sent(self): + _auth() + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "a note"]) + assert result.exit_code == 0, result.output + assert w.context.ingest_context.call_args.args[0] == [{"text": "a note", "enrich": True, "upsert": True}] + + def test_no_infer_no_enrich_and_no_upsert(self): + _auth() + for flag, field in (("--no-infer", "enrich"), ("--no-enrich", "enrich"), ("--no-upsert", "upsert")): + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "a note", flag]) + assert result.exit_code == 0, result.output + assert w.context.ingest_context.call_args.args[0][0][field] is False, flag + + def test_source_id_is_the_context_id_on_a_unified_database(self): + _auth() + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--source-id", "abc"]) + assert result.exit_code == 0, result.output + assert w.context.ingest_context.call_args.args[0][0]["context_id"] == "abc" + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--source-id", "abc", "--context-id", "abc"]) + assert result.exit_code == 0, result.output + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--source-id", "abc", "--context-id", "def"]) + assert result.exit_code != 0 + w.context.ingest_context.assert_not_called() + + def test_json_prints_the_202_verbatim(self): + _auth() + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke(app, ["--output", "json", "ingest", "--text", "a note"]) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == INGEST_202 + + def test_a_failed_item_is_reported(self): + _auth() + failed = { + **INGEST_202, + "results": [ + {"source_id": "big-1", "status": "failed", "error": "too large", "error_code": "TEXT_TOO_LARGE"} + ], + "success_count": 0, + "failed_count": 1, + } + w = _mock("unified", **{"context.ingest_context": failed}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "a note"], env=_WIDE) + # The server's own error text is still shown, and the exit code now + # says the ingest did not succeed. + assert result.exit_code != 0, result.output + out = _plain(result.output) + assert "0 success, 1 failed" in out + assert "Context ID: big-1 (failed)" in out + assert "Error: too large (TEXT_TOO_LARGE)" in out + + def test_a_failed_row_under_a_zeroed_count_still_exits_nonzero(self): + _auth() + lying = { + **INGEST_202, + "results": [ + {"source_id": "big-1", "status": "failed", "error": "too large", "error_code": "TEXT_TOO_LARGE"} + ], + "success_count": 1, + "failed_count": 0, + } + w = _mock("unified", **{"context.ingest_context": lying}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "a note"], env=_WIDE) + assert result.exit_code != 0, result.output + assert "Error: too large" in _plain(result.output) + + def test_files_are_refused(self, tmp_path): + _auth() + f = tmp_path / "doc.txt" + f.write_text("hello") + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", str(f)]) + assert result.exit_code != 0 + assert "unified" in result.output and "--text" in result.output + w.context.ingest_many.assert_not_called() + w.context.ingest_context.assert_not_called() + + def test_kind_user_name_and_markdown_are_refused(self): + _auth() + for extra in (["--kind", "memory"], ["--kind", "knowledge"], ["--user-name", "ada"], ["--markdown"]): + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", *extra]) + assert result.exit_code != 0, extra + assert "unified" in result.output + w.context.ingest.assert_not_called() + w.context.ingest_context.assert_not_called() + + def test_text_and_conversation_together_are_refused(self, tmp_path): + _auth() + f = tmp_path / "conv.json" + f.write_text(json.dumps([{"role": "user", "content": "hi"}])) + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--conversation-file", str(f)]) + assert result.exit_code != 0 + assert "exactly one" in result.output + w.context.ingest_context.assert_not_called() + + def test_a_bad_conversation_is_named_by_turn(self, tmp_path): + _auth() + cases = [ + ([{"role": "bot", "content": "x"}], "conversation[0].role"), + ([{"role": "user", "content": "x"}, {"role": "user", "content": ""}], "conversation[1].content"), + ([{"role": "user", "content": "x", "extra": 1}], "unknown field"), + ([{"role": "user", "content": "x", "name": ""}], "conversation[0].name"), + (["not an object"], "conversation[0] must be an object"), + ([], "non-empty JSON list"), + ({"role": "user"}, "non-empty JSON list"), + ] + for turns, message in cases: + f = tmp_path / "conv.json" + f.write_text(json.dumps(turns)) + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--conversation-file", str(f)]) + assert result.exit_code != 0, turns + assert message in result.output, (turns, result.output) + w.context.ingest_context.assert_not_called() + f.write_text("not json") + with _patch(_mock("unified")): + result = runner.invoke(app, ["ingest", "--conversation-file", str(f)]) + assert result.exit_code != 0 and "JSON list" in result.output + with _patch(_mock("unified")): + result = runner.invoke(app, ["ingest", "--conversation-file", str(tmp_path / "missing.json")]) + assert result.exit_code != 0 and "not found" in result.output + + def test_a_bad_happened_at_is_refused(self): + _auth() + for value in ("2026-7-1", "2026-13-40", "yesterday", "2026-07-29T10:00:00Z"): + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--happened-at", value]) + assert result.exit_code != 0, value + assert "YYYY-MM-DD" in result.output + w.context.ingest_context.assert_not_called() + + def test_attributes_must_be_json_objects(self): + _auth() + for extra in (["--attributes", "[1]"], ["--attributes", "not json"], ["--custom-attributes", '"str"']): + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", *extra]) + assert result.exit_code != 0, extra + assert "JSON object" in result.output + w.context.ingest_context.assert_not_called() + + def test_a_bad_category_is_refused(self): + _auth() + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--category", "wisdom"]) + assert result.exit_code != 0 + assert "--category must be one of" in result.output + w.context.ingest_context.assert_not_called() + + def test_forceful_relation_ids_are_trimmed_and_deduplicated(self): + _auth() + w = _mock("unified", **{"context.ingest_context": INGEST_202}) + with _patch(w): + result = runner.invoke( + app, + [ + "ingest", + "--text", + "x", + "--forceful-relation", + " a ", + "--forceful-relation", + "b", + "--forceful-relation", + "a", + ], + ) + assert result.exit_code == 0, result.output + assert w.context.ingest_context.call_args.args[0][0]["forceful_relations"] == {"ids": ["a", "b"]} + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", "--forceful-relation", " "]) + assert result.exit_code != 0 + w.context.ingest_context.assert_not_called() + + def test_deprecated_write_aliases_are_refused_on_a_unified_database(self, tmp_path): + _auth() + f = tmp_path / "doc.txt" + f.write_text("hello") + for argv in ( + ["memories", "add", "--text", "x"], + ["knowledge", "upload-text", "--text", "x"], + ["knowledge", "upload", str(f)], + ): + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, argv) + assert result.exit_code != 0, argv + assert "unified" in result.output + w.context.ingest.assert_not_called() + w.context.ingest_many.assert_not_called() + + def test_unified_only_options_are_refused_on_a_split_database(self, tmp_path): + _auth() + f = tmp_path / "conv.json" + f.write_text(json.dumps([{"role": "user", "content": "hi"}])) + for extra in ( + ["--conversation-file", str(f)], + ["--context-id", "c"], + ["--no-enrich"], + ["--instructions", "i"], + ["--happened-at", "2026-07-29"], + ["--attributes", "{}"], + ["--custom-attributes", "{}"], + ["--category", "auto"], + ["--forceful-relation", "x"], + ["--acl", "a@x.com"], + ): + w = _mock("split") + with _patch(w): + result = runner.invoke(app, ["ingest", "--text", "x", *extra]) + assert result.exit_code != 0, extra + assert "unified databases only" in result.output + assert extra[0] in result.output + w.context.ingest.assert_not_called() + w.context.ingest_context.assert_not_called() + + def test_split_ingest_is_unchanged(self): + _auth() + w = _mock( + "split", + **{"context.ingest": {"success_count": 1, "failed_count": 0, "results": [{"id": "src_1", "status": "ok"}]}}, + ) + with _patch(w): + result = runner.invoke( + app, + [ + "ingest", + "--text", + "User prefers dark mode", + "--title", + "T", + "--source-id", + "s1", + "--user-name", + "ada", + "--markdown", + "--no-infer", + "--no-upsert", + ], + ) + assert result.exit_code == 0, result.output + assert w.context.ingest.call_args.kwargs == { + "kind": "memory", + "text": "User prefers dark mode", + "title": "T", + "source_id": "s1", + "user_name": "ada", + "infer": False, + "is_markdown": True, + "upsert": False, + "database": "t1", + "collection": None, + } + w.context.ingest_context.assert_not_called() + # and knowledge text + w = _mock("split", **{"context.ingest": {"results": [{"id": "k1"}]}}) + with _patch(w): + result = runner.invoke(app, ["ingest", "--kind", "knowledge", "--text", "notes"]) + assert result.exit_code == 0, result.output + assert w.context.ingest.call_args.kwargs["kind"] == "knowledge" + + +# ── list / delete / relations / subgraph / inspect ─────────────────────────── + + +class TestUnifiedReadAndDeleteCommands: + def test_list_sends_no_kind(self): + _auth() + w = _mock("unified", **{"context.list": {"sources": [{"id": "s1", "title": "Report"}], "total": 1}}) + with _patch(w): + result = runner.invoke(app, ["list"]) + assert result.exit_code == 0, result.output + assert w.context.list.call_args.kwargs["kind"] is None + with _patch(w): + result = runner.invoke(app, ["list", "--kind", "memory"]) + assert result.exit_code != 0 and "unified" in result.output + + def test_delete_sends_no_kind(self): + _auth() + w = _mock("unified", **{"context.delete": {"success": True, "deleted_count": 1}}) + with _patch(w): + result = runner.invoke(app, ["delete", "item-1", "--yes"]) + assert result.exit_code == 0, result.output + assert w.context.delete.call_args.kwargs["kind"] is None + assert "item(s)" in result.output + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["delete", "item-1", "--kind", "knowledge", "--yes"]) + assert result.exit_code != 0 and "unified" in result.output + w.context.delete.assert_not_called() + + def test_delete_keeps_the_split_default(self): + _auth() + w = _mock("split", **{"context.delete": {"success": True, "deleted_count": 1}}) + with _patch(w): + result = runner.invoke(app, ["delete", "src_9", "--yes"]) + assert result.exit_code == 0, result.output + assert w.context.delete.call_args.kwargs["kind"] == "knowledge" + assert "knowledge source(s)" in result.output + with _patch(w): + runner.invoke(app, ["delete", "mem_1", "--kind", "memory", "--yes"]) + assert w.context.delete.call_args.kwargs["kind"] == "memory" + + def test_relations_sends_no_kind(self): + _auth() + w = _mock("unified", **{"context.relations": {"relations": []}}) + with _patch(w): + result = runner.invoke(app, ["relations", "src_1"]) + assert result.exit_code == 0, result.output + assert w.context.relations.call_args.kwargs["kind"] is None + with _patch(w): + result = runner.invoke(app, ["relations", "src_1", "--kind", "knowledge"]) + assert result.exit_code != 0 and "unified" in result.output + + def test_subgraph_sends_no_kind(self): + _auth() + w = _mock("unified", **{"context.subgraph": {"sources": []}}) + with _patch(w): + result = runner.invoke(app, ["subgraph", "src_1"]) + assert result.exit_code == 0, result.output + assert w.context.subgraph.call_args.kwargs["kind"] is None + with _patch(w): + result = runner.invoke(app, ["subgraph", "src_1", "--kind", "memory"]) + assert result.exit_code != 0 and "unified" in result.output + + def test_inspect_sends_no_kind(self): + _auth() + w = _mock("unified", **{"context.inspect": {"content": "Full text", "content_type": "text/plain"}}) + with _patch(w): + result = runner.invoke(app, ["inspect", "src_1"]) + assert result.exit_code == 0, result.output + kwargs = w.context.inspect.call_args.kwargs + assert "kind" not in kwargs and "type" not in kwargs + + def test_deprecated_memories_list_is_refused_on_a_unified_database(self): + _auth() + w = _mock("unified") + with _patch(w): + result = runner.invoke(app, ["memories", "list"]) + assert result.exit_code != 0 and "unified" in result.output + w.context.list.assert_not_called() + + +# ── hydradb database create --type / list ──────────────────────────────────── + + +class TestDatabaseLayoutCommands: + def test_create_with_type_unified(self): + _auth() + w = _mock("split", **{"databases.create": {"success": True}}) + with _patch(w): + result = runner.invoke(app, ["database", "create", "new-db", "--type", "unified"]) + assert result.exit_code == 0, result.output + assert w.databases.create.call_args.kwargs == {"database": "new-db", "layout": "unified"} + assert "unified" in result.output + + def test_create_without_type_sends_no_layout(self): + _auth() + w = _mock("split", **{"databases.create": {"success": True}}) + with _patch(w): + result = runner.invoke(app, ["database", "create", "new-db"]) + assert result.exit_code == 0, result.output + assert w.databases.create.call_args.kwargs["layout"] is None + + def test_create_rejects_an_unknown_type(self): + _auth() + w = _mock("split") + with _patch(w): + result = runner.invoke(app, ["database", "create", "new-db", "--type", "hybrid"]) + assert result.exit_code != 0 + assert "split" in result.output and "unified" in result.output + w.databases.create.assert_not_called() + + def test_wrapper_create_with_a_layout_posts_type_over_the_raw_path(self, monkeypatch): + capture = _capture_post(monkeypatch, body={"success": True, "data": {"status": "accepted"}, "meta": {}}) + out = _real_wrapper(_sdk_500).databases.create(database="new", layout="unified") + assert out == {"status": "accepted"} + assert capture.calls[0]["url"] == "http://test.local/databases" + assert capture.calls[0]["json"] == {"database": "new", "type": "unified"} + + def test_wrapper_create_without_a_layout_is_the_sdk_call(self): + seen = {} + + def handler(request): + seen["path"] = request.url.path + seen["body"] = json.loads(request.content) + return httpx.Response(200, json={"success": True, "data": {"status": "accepted"}, "meta": {}}) + + _real_wrapper(handler).databases.create(database="new") + assert seen["path"] == "/databases" + assert "type" not in seen["body"] + + def test_wrapper_create_rejects_an_unknown_layout(self): + with pytest.raises(ValueError): + _real_wrapper(_sdk_500).databases.create(database="new", layout="hybrid") + + def test_list_shows_each_layout(self): + _auth() + w = _mock( + "split", + **{ + "databases.list": { + "databases": ["a", "b"], + "details": [{"database": "a", "type": "unified"}, {"database": "b", "type": "split"}], + } + }, + ) + with _patch(w): + result = runner.invoke(app, ["database", "list"], env=_WIDE) + assert result.exit_code == 0, result.output + lines = _lines(result) + assert any("Type" in line for line in lines) + assert any("a" in line.split("│") or " a " in line for line in lines if "unified" in line), lines + assert any(" b " in line for line in lines if "split" in line), lines + + def test_list_without_details_shows_split(self): + _auth() + w = _mock("split", **{"databases.list": {"databases": ["a"]}}) + with _patch(w): + result = runner.invoke(app, ["database", "list"], env=_WIDE) + assert result.exit_code == 0, result.output + assert any(" a " in line and "split" in line for line in _lines(result))