From df436bc8d9bc529dd3b1afeabfebd567491dd295 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 04:37:24 +0000 Subject: [PATCH 1/7] test(kv_config): cover the KV config/secret seam (49% -> 100%) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `kv_config.py` is the KV seam the cost/routing hub and `credentials.py` read config and provider secrets through (never `os.getenv` at runtime), but its own surface was only exercised incidentally by other suites — line coverage sat at 49%, leaving the secret sub-surface and the pg_llm_batch adapter untested. Add a focused `tests/test_kv_config.py` (13 tests, dependency-free — small fakes stand in for the pg_llm_batch config/secret stores) pinning the full contract: - InMemoryConfigStore: seed loading, get/set roundtrip + default, get_category returns a non-aliasing copy, show_config sorted ordering, and the set_secret/get_secret/require_secret surface (incl. secrets staying out of show_config and require_secret raising KeyError when absent). - PostgresConfigStoreAdapter: get/set delegation, and get_secret/require_secret behavior with and without a backing secret store (default vs. KeyError, and swallowing a backing-store error to the default). - get_config_store: the no-DSN in-memory selection path, seeded and unseeded. kv_config.py line coverage 49% -> 100% (the pg_llm_batch-present branch remains `# pragma: no cover`, as before). Full suite: 313 passed (was 300). Test-only; no production behavior changes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_kv_config.py | 170 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 tests/test_kv_config.py diff --git a/tests/test_kv_config.py b/tests/test_kv_config.py new file mode 100644 index 000000000..072d39813 --- /dev/null +++ b/tests/test_kv_config.py @@ -0,0 +1,170 @@ +"""KV config seam: in-memory store surface, the pg_llm_batch adapter, and the +``get_config_store`` selector. + +These run entirely on the dependency-free in-memory path plus small fakes for +the ``pg_llm_batch`` config/secret stores — no Postgres or ``pg_llm_batch`` +install is needed. They pin the KV contract (``get``/``set``/``show_config`` + +the secret sub-surface) that ``credentials.py`` and the cost/routing hub read +config and provider secrets through, never ``os.getenv`` at runtime. +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator.kv_config import ( # noqa: E402 + InMemoryConfigStore, + PostgresConfigStoreAdapter, + get_config_store, +) + + +# --- InMemoryConfigStore --------------------------------------------------- + + +def test_in_memory_seed_is_loaded_via_set() -> None: + """A ``seed`` mapping is materialised through ``set`` at construction.""" + store = InMemoryConfigStore(seed={"pricing": {"openai_input": 1.25}}) + assert store.get("pricing", "openai_input") == 1.25 + + +def test_in_memory_get_returns_default_when_unset() -> None: + """Reads of an absent category/key fall back to the supplied default.""" + store = InMemoryConfigStore() + assert store.get("missing", "key") is None + assert store.get("missing", "key", "fallback") == "fallback" + + +def test_in_memory_set_then_get_roundtrips() -> None: + """A value written under a category/key is read back unchanged.""" + store = InMemoryConfigStore() + store.set("routing", "batch_threshold", 42) + assert store.get("routing", "batch_threshold") == 42 + + +def test_in_memory_get_category_returns_a_copy() -> None: + """``get_category`` returns a snapshot dict that does not alias internal state.""" + store = InMemoryConfigStore(seed={"routing": {"a": 1, "b": 2}}) + category = store.get_category("routing") + assert category == {"a": 1, "b": 2} + category["a"] = 999 + assert store.get("routing", "a") == 1 # mutation of the copy does not leak back + assert store.get_category("absent") == {} + + +def test_in_memory_show_config_is_sorted() -> None: + """``show_config`` yields every entry ordered by category then key.""" + store = InMemoryConfigStore() + store.set("z_cat", "k2", "v2") + store.set("z_cat", "k1", "v1") + store.set("a_cat", "k", "v") + assert list(store.show_config()) == [ + ("a_cat", "k", "v"), + ("z_cat", "k1", "v1"), + ("z_cat", "k2", "v2"), + ] + + +def test_in_memory_secret_surface() -> None: + """Secrets round-trip via ``set_secret``/``get_secret``/``require_secret`` and + stay out of ``show_config``.""" + store = InMemoryConfigStore() + store.set_secret("OPENAI_API_KEY", "sk-secret") + assert store.get_secret("OPENAI_API_KEY") == "sk-secret" + assert store.get_secret("absent") is None + assert store.get_secret("absent", "dflt") == "dflt" + assert store.require_secret("OPENAI_API_KEY") == "sk-secret" + # A secret is never surfaced by the plain-config listing. + assert all(name != "OPENAI_API_KEY" for _cat, name, _val in store.show_config()) + + +def test_in_memory_require_secret_raises_when_absent() -> None: + """``require_secret`` raises ``KeyError`` for an unconfigured secret.""" + store = InMemoryConfigStore() + with pytest.raises(KeyError): + store.require_secret("MISSING") + + +# --- PostgresConfigStoreAdapter (over fakes) ------------------------------- + + +class _FakeConfig: + """Minimal stand-in for ``pg_llm_batch.PostgresConfigStore``.""" + + def __init__(self) -> None: + self.tree: dict[tuple[str, str], object] = {} + + def get(self, category: str, key: str, default: object = None) -> object: + return self.tree.get((category, key), default) + + def set(self, category: str, key: str, value: object) -> None: + self.tree[(category, key)] = value + + +class _FakeSecret: + """Minimal stand-in for ``pg_llm_batch.SecretStore``.""" + + def __init__(self, secrets: dict[str, str]) -> None: + self._secrets = secrets + + def require_secret(self, name: str) -> str: + return self._secrets[name] # raises KeyError when absent + + +def test_adapter_delegates_get_and_set() -> None: + """The adapter forwards config reads/writes to the backing store.""" + backing = _FakeConfig() + adapter = PostgresConfigStoreAdapter(backing) + adapter.set("pricing", "input", 3.0) + assert backing.tree[("pricing", "input")] == 3.0 + assert adapter.get("pricing", "input") == 3.0 + assert adapter.get("pricing", "missing", "dflt") == "dflt" + + +def test_adapter_get_secret_without_secret_store_returns_default() -> None: + """With no secret store, ``get_secret`` returns the default rather than raising.""" + adapter = PostgresConfigStoreAdapter(_FakeConfig(), secret_store=None) + assert adapter.get_secret("OPENAI_API_KEY") is None + assert adapter.get_secret("OPENAI_API_KEY", "dflt") == "dflt" + + +def test_adapter_get_secret_delegates_and_swallows_errors() -> None: + """``get_secret`` returns a configured secret, and degrades to the default + when the backing store raises (e.g. secret absent).""" + adapter = PostgresConfigStoreAdapter( + _FakeConfig(), secret_store=_FakeSecret({"OPENAI_API_KEY": "sk-x"}) + ) + assert adapter.get_secret("OPENAI_API_KEY") == "sk-x" + assert adapter.get_secret("ABSENT", "dflt") == "dflt" + + +def test_adapter_require_secret_without_store_raises() -> None: + """``require_secret`` raises ``KeyError`` when no secret store is attached.""" + adapter = PostgresConfigStoreAdapter(_FakeConfig(), secret_store=None) + with pytest.raises(KeyError): + adapter.require_secret("OPENAI_API_KEY") + + +def test_adapter_require_secret_delegates() -> None: + """``require_secret`` returns the backing store's secret when present.""" + adapter = PostgresConfigStoreAdapter( + _FakeConfig(), secret_store=_FakeSecret({"OPENAI_API_KEY": "sk-y"}) + ) + assert adapter.require_secret("OPENAI_API_KEY") == "sk-y" + + +# --- get_config_store selector -------------------------------------------- + + +def test_get_config_store_without_dsn_is_in_memory() -> None: + """No DSN yields the dependency-free in-memory store, seeded when asked.""" + store = get_config_store() + assert isinstance(store, InMemoryConfigStore) + seeded = get_config_store(seed={"routing": {"batch_threshold": 7}}) + assert isinstance(seeded, InMemoryConfigStore) + assert seeded.get("routing", "batch_threshold") == 7 From 78ab40202c2f43b207267d1e4f7595fb887e31bd Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 04:51:06 +0000 Subject: [PATCH 2/7] test(credentials): cover the Postgres backend bootstrap surface (83% -> 100%) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the KV credential-seam coverage started in the kv_config test: the non-DB surface of credentials.py was uncovered — PostgresCredentialBackend argument validation, from_env bootstrap-transport reads, and the _select_backend postgres branch (the live pgcrypto/psycopg methods stay # pragma: no cover). Add tests/test_credentials_backend.py (8 tests, no Postgres needed) pinning the "KV, not env" bootstrap boundary: - PostgresCredentialBackend raises NotConfigured on an empty DSN or passphrase, and stores both (lazy schema) when valid. - from_env builds from exactly the two bootstrap env vars, and fails loudly when the DSN is unset (no silent empty backend). - _select_backend: memory default (case-insensitive), the postgres branch routes through from_env, and an unknown selector raises. credentials.py 83% -> 100%; combined with kv_config.py the KV seam is now 100%. Full suite 321 passed (was 313). Test-only; no production behavior changes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_credentials_backend.py | 94 +++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) create mode 100644 tests/test_credentials_backend.py diff --git a/tests/test_credentials_backend.py b/tests/test_credentials_backend.py new file mode 100644 index 000000000..52c1057bc --- /dev/null +++ b/tests/test_credentials_backend.py @@ -0,0 +1,94 @@ +"""Postgres credential-backend construction + bootstrap selection. + +Covers the non-DB surface of ``credentials.py``: ``PostgresCredentialBackend`` +argument validation, ``from_env`` bootstrap-transport reads, and the +``_select_backend`` postgres branch. The live pgcrypto/psycopg methods +(``_connect``/``get``/``set``) require a real Postgres and stay +``# pragma: no cover``. No Postgres is needed for any test here. + +These pin the org "KV, not env" invariant at its bootstrap boundary: the DSN +and passphrase are the only permitted environment reads, and a missing one must +fail loudly (``NotConfigured``) rather than degrade to a silent/empty backend. +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator.credentials import ( # noqa: E402 + InMemoryCredentialBackend, + NotConfigured, + PostgresCredentialBackend, + _select_backend, +) + +_DSN_ENV = "CONTEXTUAL_ORCHESTRATOR_KV_DSN" +_PASS_ENV = "CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE" +_BACKEND_ENV = "CONTEXTUAL_ORCHESTRATOR_KV_BACKEND" + + +def test_postgres_backend_requires_dsn() -> None: + """An empty bootstrap DSN fails loudly instead of building a broken backend.""" + with pytest.raises(NotConfigured): + PostgresCredentialBackend("", "passphrase") + + +def test_postgres_backend_requires_passphrase() -> None: + """An empty bootstrap passphrase fails loudly.""" + with pytest.raises(NotConfigured): + PostgresCredentialBackend("postgresql://localhost/db", "") + + +def test_postgres_backend_stores_bootstrap_transport() -> None: + """A valid DSN + passphrase construct a backend that has not yet touched the DB.""" + backend = PostgresCredentialBackend("postgresql://localhost/db", "s3cret") + assert backend._dsn == "postgresql://localhost/db" + assert backend._passphrase == "s3cret" + assert backend._ensured is False # schema is ensured lazily on first connect + + +def test_from_env_reads_only_bootstrap_vars(monkeypatch: pytest.MonkeyPatch) -> None: + """``from_env`` builds the backend from the two bootstrap-transport env vars.""" + monkeypatch.setenv(_DSN_ENV, "postgresql://kv-host/registry") + monkeypatch.setenv(_PASS_ENV, "unlock-phrase") + backend = PostgresCredentialBackend.from_env() + assert isinstance(backend, PostgresCredentialBackend) + assert backend._dsn == "postgresql://kv-host/registry" + assert backend._passphrase == "unlock-phrase" + + +def test_from_env_missing_dsn_fails_loudly(monkeypatch: pytest.MonkeyPatch) -> None: + """With the DSN env unset, ``from_env`` surfaces ``NotConfigured`` (no silent empty backend).""" + monkeypatch.delenv(_DSN_ENV, raising=False) + monkeypatch.setenv(_PASS_ENV, "unlock-phrase") + with pytest.raises(NotConfigured): + PostgresCredentialBackend.from_env() + + +def test_select_backend_defaults_to_memory(monkeypatch: pytest.MonkeyPatch) -> None: + """Unset/`memory` selector yields the dependency-free in-memory backend.""" + monkeypatch.delenv(_BACKEND_ENV, raising=False) + assert isinstance(_select_backend(), InMemoryCredentialBackend) + monkeypatch.setenv(_BACKEND_ENV, "MEMORY") # case-insensitive + assert isinstance(_select_backend(), InMemoryCredentialBackend) + + +def test_select_backend_postgres_branch(monkeypatch: pytest.MonkeyPatch) -> None: + """The ``postgres`` selector routes through ``from_env`` to a Postgres backend.""" + monkeypatch.setenv(_BACKEND_ENV, "postgres") + monkeypatch.setenv(_DSN_ENV, "postgresql://kv-host/registry") + monkeypatch.setenv(_PASS_ENV, "unlock-phrase") + backend = _select_backend() + assert isinstance(backend, PostgresCredentialBackend) + + +def test_select_backend_unknown_selector_raises(monkeypatch: pytest.MonkeyPatch) -> None: + """An unrecognized selector fails loudly rather than guessing a backend.""" + monkeypatch.setenv(_BACKEND_ENV, "vault") + with pytest.raises(NotConfigured): + _select_backend() From c0686f68caf5606a98eaffa20ea3e232b0974977 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 04:57:10 +0000 Subject: [PATCH 3/7] test(token_counting): cover heuristic edges, pg adapter, selector (74% -> 100%) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cost hub's token-accounting seam was 74% covered. Add tests/test_token_counting.py (5 tests, dependency-free — a fake stands in for pg_llm_batch.TokenCounter): the heuristic counter's empty/whitespace-only zero path and word/punctuation monotonicity; PgTiktokenAdapter count_text/count_messages delegation (incl. the non-dict message -> "" branch); and build_token_counter's no-DSN heuristic selection (the pg_llm_batch import path stays # pragma: no cover). token_counting.py 74% -> 100%. Full suite 326 passed (was 300). Test-only. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_token_counting.py | 72 ++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 tests/test_token_counting.py diff --git a/tests/test_token_counting.py b/tests/test_token_counting.py new file mode 100644 index 000000000..da8222c3e --- /dev/null +++ b/tests/test_token_counting.py @@ -0,0 +1,72 @@ +"""Token-counting seam: heuristic estimator edge cases, the pg_tiktoken adapter, +and the ``build_token_counter`` selector. + +Dependency-free — a small fake stands in for ``pg_llm_batch.TokenCounter`` so the +adapter delegation is exercised without Postgres/pg_tiktoken (the real +pg_llm_batch import path stays ``# pragma: no cover``). These pin the cost hub's +token accounting: the heuristic is deterministic (tests can assert on it), and +the selector never reads the environment. +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator.token_counting import ( # noqa: E402 + HeuristicTokenCounter, + PgTiktokenAdapter, + build_token_counter, +) + + +def test_heuristic_empty_and_whitespace_only_are_zero() -> None: + """Empty text, and text with no word units (whitespace only), count as 0.""" + counter = HeuristicTokenCounter() + assert counter.count_text("") == 0 + assert counter.count_text(" \t\n ") == 0 # no \w or punctuation units + + +def test_heuristic_counts_words_and_punctuation_monotonically() -> None: + """Counting is >=1 for non-empty content and grows with more units.""" + counter = HeuristicTokenCounter() + one = counter.count_text("hello") + more = counter.count_text("hello, world!") + assert one >= 1 + assert more > one + + +class _FakePgCounter: + """Stand-in for ``pg_llm_batch.TokenCounter`` — records calls, returns a fixed count.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, str]] = [] + + def count_tokens(self, text: str, model: str) -> int: + self.calls.append((text, model)) + return 7 + + +def test_pg_adapter_count_text_delegates() -> None: + """``PgTiktokenAdapter.count_text`` forwards to the backing pg counter.""" + fake = _FakePgCounter() + adapter = PgTiktokenAdapter(fake) + assert adapter.count_text("hello", "gpt-x") == 7 + assert fake.calls == [("hello", "gpt-x")] + + +def test_pg_adapter_count_messages_sums_per_message() -> None: + """``PgTiktokenAdapter.count_messages`` sums per-message counts (non-dict → empty).""" + fake = _FakePgCounter() + adapter = PgTiktokenAdapter(fake) + total = adapter.count_messages([{"content": "a"}, {"content": "b"}, "not-a-dict"], "m") + assert total == 21 # 3 messages * fixed 7 + assert fake.calls == [("a", "m"), ("b", "m"), ("", "m")] + + +def test_build_token_counter_without_dsn_is_heuristic() -> None: + """With no DSN, the dependency-free heuristic counter is selected.""" + assert isinstance(build_token_counter(), HeuristicTokenCounter) + assert isinstance(build_token_counter(postgres_dsn=None), HeuristicTokenCounter) From 6d4ba405a1ca646dca22ec0a083b17c1ba1fd767 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 06:00:36 +0000 Subject: [PATCH 4/7] test(batch_routing): cover embeddings batch backends + helpers (81% -> 100%) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit batch_routing is the cost hub's sync-vs-batch routing/execution surface; its embeddings path was undercovered. Add tests/test_batch_routing_embeddings.py (9 tests, no network / no pg-llm-batch install — a fake async client + fake assembler stand in): - LocalEmbeddingBatchBackend submit/poll/retrieve incl. the dependency-free token-count fallback (no token_counter) - PgLlmBatchEmbeddingBackend submit/poll/retrieve against a fake BatchAPIClient (both assembler and memory:// payload paths), result ordering + usage mapping, and empty-on-download-failure - helpers: heuristic_embedding (dimension + positive-guard), _extract_embedding (parse/defaults), _extract_answer (empty choices), cheapest_upstream (empty candidates -> None), EmbeddingBatchRequest.to_jsonl_line + build_embeddings_jsonl_body batch_routing.py 81% -> 100%. Test-only; no production change. Verified: 9 passed; full suite green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_batch_routing_embeddings.py | 118 +++++++++++++++++++++++++ 1 file changed, 118 insertions(+) create mode 100644 tests/test_batch_routing_embeddings.py diff --git a/tests/test_batch_routing_embeddings.py b/tests/test_batch_routing_embeddings.py new file mode 100644 index 000000000..530b659e0 --- /dev/null +++ b/tests/test_batch_routing_embeddings.py @@ -0,0 +1,118 @@ +"""Embeddings batch-routing coverage: local + pg-llm-batch backends and helpers. + +Exercises the cost-hub's embeddings batch path (the repo's sync-vs-batch routing +role): the in-process ``LocalEmbeddingBatchBackend`` (incl. the dependency-free +token-count fallback), the ``PgLlmBatchEmbeddingBackend`` submit/poll/retrieve +flow against a fake async client, and the small helpers (``heuristic_embedding``, +``_extract_embedding``, ``_extract_answer``, ``cheapest_upstream``, +``build_embeddings_jsonl_body``). No network or pg-llm-batch install needed. +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import batch_routing as b # noqa: E402 + + +# --- small helpers ------------------------------------------------------- + + +def test_cheapest_upstream_none_for_empty_candidates() -> None: + assert b.cheapest_upstream([], price_book={}) is None + + +def test_extract_answer_empty_choices_is_empty_string() -> None: + assert b._extract_answer({"choices": []}) == "" + assert b._extract_answer({"choices": [{"message": {"content": "hi"}}]}) == "hi" + + +def test_heuristic_embedding_dimension_and_guard() -> None: + vec = b.heuristic_embedding("고객 이탈", dimension=6) + assert len(vec) == 6 + with pytest.raises(ValueError): + b.heuristic_embedding("x", dimension=0) + + +def test_extract_embedding_parses_and_defaults() -> None: + assert b._extract_embedding({"data": [{"embedding": [1, 2, 3]}]}) == [1.0, 2.0, 3.0] + assert b._extract_embedding({}) == [] + assert b._extract_embedding({"data": []}) == [] + + +def test_embedding_request_to_jsonl_line_and_body() -> None: + req = b.EmbeddingBatchRequest("hello world", custom_id="c1") + line = req.to_jsonl_line() + assert line["custom_id"] == "c1" + body = b.build_embeddings_jsonl_body([req]) + assert "c1" in body and body.startswith("{") + + +# --- LocalEmbeddingBatchBackend (token-count fallback) ------------------- + + +def test_local_embedding_backend_roundtrip_with_token_fallback() -> None: + """No token_counter -> the word-count fallback is used; submit/poll/retrieve work.""" + backend = b.LocalEmbeddingBatchBackend(dimension=8) # token_counter=None -> fallback + job = backend.submit([b.EmbeddingBatchRequest("two words", custom_id="c1")]) + assert backend.poll(job)["is_complete"] is True + results = backend.retrieve(job) + assert len(results) == 1 and len(results[0].embedding) == 8 + + +# --- PgLlmBatchEmbeddingBackend (fake async client) ---------------------- + + +class _FakeClient: + async def upload_jsonl(self, path, alias): + return {"id": "file-1"} + + async def create_batch_job(self, input_file_id, alias, *, endpoint, metadata=None): + return {"id": "batch-1", "status": "validating"} + + async def get_batch_status(self, job_id, alias): + return {"status": "completed", "is_complete": True, "progress_percentage": 100} + + async def download_results(self, job_id, alias): + return { + "success": True, + "responses": [ + { + "custom_id": "c1", + "response": {"body": {"data": [{"embedding": [0.1, 0.2]}], "usage": {"prompt_tokens": 3}}}, + } + ], + } + + +class _FakeAssembler: + def assemble(self, lines): + return "file://assembled" + + +@pytest.mark.parametrize("assembler", [None, _FakeAssembler()]) +def test_pg_embedding_backend_submit_poll_retrieve(assembler) -> None: + backend = b.PgLlmBatchEmbeddingBackend(_FakeClient(), payload_assembler=assembler) + job = backend.submit([b.EmbeddingBatchRequest("hi", custom_id="c1")]) + assert job.job_id == "batch-1" and job.request_count == 1 + status = backend.poll(job) + assert status["is_complete"] is True + results = backend.retrieve(job) + assert len(results) == 1 + assert results[0].custom_id == "c1" and results[0].embedding == [0.1, 0.2] + assert results[0].prompt_tokens == 3 + + +def test_pg_embedding_backend_retrieve_empty_on_failure() -> None: + class _FailClient(_FakeClient): + async def download_results(self, job_id, alias): + return {"success": False} + + backend = b.PgLlmBatchEmbeddingBackend(_FailClient()) + job = backend.submit([b.EmbeddingBatchRequest("hi", custom_id="c1")]) + assert backend.retrieve(job) == [] From f4847f9139c79494160a07e74f29abff857372e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 06:57:19 +0000 Subject: [PATCH 5/7] test(cost-router): cover defensive edge branches (89%->96%) Add targeted tests for previously-uncovered defensive/edge paths in the cost-hub router, all reachable and behavior-pinning: - _provider_from_base_url: mock scheme, real host extraction, empty input, and a malformed URL that must fall through the guarded parse to "" (never raise). - _positive_int: valid parse, ValueError/TypeError fallbacks, non-positive -> default. - _weighted_average_embedding: empty parts, all-empty-vector parts, and the weighted mean. - poll_batch / retrieve_batch / embeddings_batch_document: KeyError on an unknown job id. - _split_embedding_input / _force_token_safe_chunks: empty input, over-max_chars fixed-width split, and the token-dense single-unit midpoint-recursion fallback. - _count_embedding_tokens: tolerates a failing token counter (word-count fallback) and coerces a non-positive count on non-empty text to 1. cost_router.py coverage 89% -> 96% (full suite 341 passed). Tests only; no production behavior changes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_cost_router.py | 83 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/tests/test_cost_router.py b/tests/test_cost_router.py index 19cf25d10..8c56b89c4 100644 --- a/tests/test_cost_router.py +++ b/tests/test_cost_router.py @@ -5,6 +5,8 @@ from pathlib import Path import sys +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contextual_orchestrator import ( # noqa: E402 @@ -19,6 +21,11 @@ TaskOrchestrator, ) from contextual_orchestrator.batch_routing import PgLlmBatchBackend # noqa: E402 +from contextual_orchestrator.cost_router import ( # noqa: E402 + _positive_int, + _provider_from_base_url, + _weighted_average_embedding, +) class _FailingLedgerStore: @@ -189,3 +196,79 @@ async def download_results(self, batch_id, endpoint_alias): _fn() print(f"ok {_name}") print("ok") + + +# --- module helpers + defensive edge branches -------------------------------- + + +def test_provider_from_base_url_variants() -> None: + """mock scheme, real host extraction, empty input, and a malformed URL that + must fall through the guarded parse to an empty string (never raise).""" + assert _provider_from_base_url("mock://a") == "mock" + assert _provider_from_base_url("https://api.openai.com/v1") == "api.openai.com" + assert _provider_from_base_url("") == "" + assert _provider_from_base_url("http://[::1") == "" # malformed -> guarded fallback + + +def test_positive_int_parses_and_falls_back_to_default() -> None: + assert _positive_int("5", 1) == 5 + assert _positive_int("x", 7) == 7 # ValueError -> default + assert _positive_int(None, 7) == 7 # TypeError -> default + assert _positive_int("-3", 9) == 9 # non-positive -> default + assert _positive_int("0", 9) == 9 + + +def test_weighted_average_embedding_edges_and_mean() -> None: + assert _weighted_average_embedding([]) == [] + assert _weighted_average_embedding([([], 3), ([], 2)]) == [] # no non-empty vectors + # (1*[1,0] + 3*[3,4]) / 4 == [2.5, 3.0] + assert _weighted_average_embedding([([1.0, 0.0], 1), ([3.0, 4.0], 3)]) == [2.5, 3.0] + + +def test_batch_and_embedding_lookups_raise_keyerror_for_unknown_ids() -> None: + coordinator = _coordinator() + with pytest.raises(KeyError): + coordinator.poll_batch("nope") + with pytest.raises(KeyError): + coordinator.retrieve_batch("nope") + with pytest.raises(KeyError): + coordinator.embeddings_batch_document("nope") + + +def test_split_embedding_input_handles_empty_and_oversize_no_whitespace() -> None: + coordinator = _coordinator() + assert coordinator._split_embedding_input("", model="m", max_tokens=8, max_chars=8) == [("", 0)] + assert coordinator._force_token_safe_chunks("", model="m", max_tokens=8, max_chars=8) == [("", 0)] + # Over max_chars -> fixed-width char split. + chunks = coordinator._force_token_safe_chunks("x" * 40, model="m", max_tokens=1000, max_chars=8) + assert len(chunks) > 1 + assert all(len(text) <= 8 for text, _ in chunks) + + # Within max_chars but token-dense and a single unit (no unit split helps) -> + # the midpoint-recursion fallback keeps splitting until each chunk fits. + class _PerCharCounter: + def count_text(self, text, model): + return len(text) # one token per character + + coordinator.token_counter = _PerCharCounter() + dense = coordinator._force_token_safe_chunks("abcdefgh", model="m", max_tokens=1, max_chars=8) + assert len(dense) == 8 + assert all(len(text) <= 1 for text, _ in dense) + + +def test_count_embedding_tokens_tolerates_counter_failure_and_zero() -> None: + coordinator = _coordinator() + + class _Raising: + def count_text(self, text, model): + raise RuntimeError("counter down") + + class _Zero: + def count_text(self, text, model): + return 0 + + coordinator.token_counter = _Raising() + assert coordinator._count_embedding_tokens("a b c", "m") == 3 # word-count fallback + coordinator.token_counter = _Zero() + assert coordinator._count_embedding_tokens("abc", "m") == 1 # non-empty but 0 -> 1 + assert coordinator._count_embedding_tokens("", "m") == 0 # empty -> 0 From 4b4eb6058c3a0e840bd2dc8326dbc3147679a91c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 09:21:23 +0000 Subject: [PATCH 6/7] test(cost-router): reach 100% line coverage (96%->100%) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close the remaining cost_router.py branches to meet the org's 100% coverage standard: - _served_provider_model: the fallback path when the trace names an agent the orchestrator cannot resolve (lookup raises -> "unknown", fallback_model). - embeddings_batch_document: the pending return while the backend poll is not yet complete; the count_text token fallback when an item reports non-positive prompt_tokens and its request carries a zero token_count; and the empty-source branch when an input receives no returned embedding item. - _weighted_average_embedding: mark the total_weight<=0 guard `# pragma: no cover` — it is unreachable (each summand is max(1, ...) over a guaranteed-non-empty parts list), documented inline. Tests added drive a fake embeddings backend (pending / complete-with-gaps) via the public submit/document surface. Full suite: 344 passed; cost_router.py 100%. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- contextual_orchestrator/cost_router.py | 2 +- tests/test_cost_router.py | 98 ++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index bfbe159db..1a0049e2e 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -632,7 +632,7 @@ def _weighted_average_embedding(parts: List[tuple[List[float], int]]) -> List[fl return [] dimension = max(len(vector) for vector in vectors) total_weight = sum(max(1, int(weight)) for _vector, weight in parts) - if total_weight <= 0: + if total_weight <= 0: # pragma: no cover - unreachable: each term is max(1, ...) over a guaranteed-non-empty parts list total_weight = len(parts) reduced: List[float] = [] for offset in range(dimension): diff --git a/tests/test_cost_router.py b/tests/test_cost_router.py index 8c56b89c4..2dda0fb44 100644 --- a/tests/test_cost_router.py +++ b/tests/test_cost_router.py @@ -272,3 +272,101 @@ def count_text(self, text, model): coordinator.token_counter = _Zero() assert coordinator._count_embedding_tokens("abc", "m") == 1 # non-empty but 0 -> 1 assert coordinator._count_embedding_tokens("", "m") == 0 # empty -> 0 + + +# --- remaining branch coverage: provider resolution + embeddings document --- + + +def test_served_provider_model_falls_back_when_agent_lookup_raises() -> None: + """A trace naming an agent the orchestrator cannot resolve falls back to + ``('unknown', fallback_model)`` instead of raising.""" + coordinator = _coordinator() + provider, model = coordinator._served_provider_model( + {"trace": [{"served_agent_id": "__no_such_agent__"}]}, "fallback-model" + ) + assert (provider, model) == ("unknown", "fallback-model") + + +class _FakeJob: + """Minimal BatchJob-shaped handle a fake embedding backend can return.""" + + +def _embedding_coordinator(backend): + """A coordinator wired to an explicit embeddings backend.""" + agents = [ + ModelAgent(id="mock_worker", model="mock-a", base_url="mock://a", provider_name="mock", + tags=("reasoning", "coding", "writing"), priority=1), + ] + orchestrator = TaskOrchestrator(agents) + config = InMemoryConfigStore() + price_book = PriceBook(config) + price_book.set_price(PriceEntry("mock", "mock-a", prompt_price_per_1k=1.0, completion_price_per_1k=2.0)) + return CostRoutingCoordinator( + orchestrator, config, price_book=price_book, embedding_batch_backend=backend + ) + + +def test_embeddings_batch_document_returns_pending_while_incomplete() -> None: + """A backend whose poll is not complete yields the pending document + (``embeddings is None``) and does not record any cost.""" + from contextual_orchestrator.batch_routing import BatchJob + + class _Pending: + def submit(self, requests, metadata=None): + return BatchJob(job_id="emb-pending", backend="fake", status="processing", + request_count=len(requests)) + + def poll(self, job): + return {"is_complete": False, "status": "processing"} + + def retrieve(self, job): # pragma: no cover - not reached while pending + return [] + + coordinator = _embedding_coordinator(_Pending()) + job = coordinator.submit_embeddings_batch(["hello world"], attribution={"team": "a"}) + doc = coordinator.embeddings_batch_document(job.job_id) + assert doc["embeddings"] is None + assert doc["status"] == "processing" + assert doc["backend"] == "fake" + assert coordinator.ledger.records() == [] + + +def test_embeddings_batch_document_token_fallback_and_empty_source() -> None: + """An item with non-positive prompt_tokens whose request has a zero + token_count triggers the count_text fallback, and a source input that + receives no returned item yields an empty embedding entry.""" + from contextual_orchestrator.batch_routing import BatchJob, EmbeddingBatchResultItem + + class _CompleteSourceZeroOnly: + def __init__(self): + self._requests = [] + + def submit(self, requests, metadata=None): + self._requests = list(requests) + return BatchJob(job_id="emb-done", backend="fake", status="completed", + request_count=len(requests)) + + def poll(self, job): + return {"is_complete": True, "status": "completed"} + + def retrieve(self, job): + # Return an item only for source 0 (the "" input -> token_count 0), + # with prompt_tokens 0 so the count_text fallback runs. Source 1 + # gets no item, so its parts list stays empty. + return [ + EmbeddingBatchResultItem( + custom_id=r.custom_id, index=i, embedding=[1.0, 2.0], + prompt_tokens=0, model=r.model, + ) + for i, r in enumerate(self._requests) + if r.source_index == 0 + ] + + coordinator = _embedding_coordinator(_CompleteSourceZeroOnly()) + job = coordinator.submit_embeddings_batch(["", "world"], attribution={"team": "a"}) + doc = coordinator.embeddings_batch_document(job.job_id) + # Two source inputs -> two embedding entries; source 1 received no item. + by_index = {entry["index"]: entry for entry in doc["embeddings"]} + assert by_index[1]["embedding"] == [] # empty-source branch + assert by_index[0]["embedding"] # source 0 produced a reduced vector + assert doc["token_counts"][1] == 0 From 2f9a4e4add83916ab0ed9fbbae873c5bbc3a0ffb Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 09:37:20 +0000 Subject: [PATCH 7/7] test(cost-ledger): reach 100% line coverage (92%->100%) Add 13 focused tests closing the remaining cost_ledger.py branches to meet the org's 100% coverage standard, all with real tests (no pragmas): - InMemoryUsageTelemetrySink event-list trim past max_events. - _emit_usage_event best-effort swallow when the sink raises. - NonBlockingLedgerStore: zero-queue-size guard, queue.Full drop path (+dropped telemetry), query delegation, flush timeout, worker success-path mark/emit. - InMemoryLedgerStore.__len__. - SQL ledger store time-window WHERE-clause builder (start/end params). - CostLedger: non-blocking store wrapping via flag; attribution passed as an AttributionDimensions instance; inline append failure marks health + emits an export_error event; flush no-op when the store has no flush(). Deterministic threading via Event-synchronized backends (no sleeps). Full suite: 357 passed; cost_ledger.py 100%. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- tests/test_cost_ledger.py | 228 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 228 insertions(+) diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index 8051712a9..af6848b01 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -5,17 +5,22 @@ from pathlib import Path import sqlite3 import sys +import threading sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contextual_orchestrator.cost_ledger import ( # noqa: E402 ATTRIBUTION_DIMENSIONS, + AttributionDimensions, CostLedger, + InMemoryLedgerStore, InMemoryUsageTelemetrySink, NonBlockingLedgerStore, PriceBook, PriceEntry, SqlLedgerStore, + UsageRecord, + UsageTelemetryEvent, dimension_catalog, ) from contextual_orchestrator.conventions import is_two_word_snake_case # noqa: E402 @@ -253,6 +258,229 @@ def test_dimension_catalog_covers_all_required_dimensions() -> None: assert names == {"account", "service", "upstream_api", "model_name", "team", "group", "company"} +class _RecordingBackend: + """Working ledger backend that records appends and returns fixed query rows.""" + + def __init__(self, rows=None) -> None: + self.rows = rows if rows is not None else [] + self.appended = [] + + def append(self, record) -> None: + self.appended.append(record) + + def query(self, start=None, end=None): + return self.rows + + +class _BlockingBackend: + """Ledger backend whose ``append`` blocks until ``release`` is set.""" + + def __init__(self) -> None: + self.started = threading.Event() + self.release = threading.Event() + self.appended = [] + + def append(self, record) -> None: + self.started.set() + self.release.wait(timeout=5.0) + self.appended.append(record) + + def query(self, start=None, end=None): + return [] + + +class _RaisingTelemetrySink: + """Telemetry sink that always raises, to exercise best-effort swallowing.""" + + def emit_usage(self, event) -> None: + raise RuntimeError("telemetry backend is down") + + +def _sample_record(usage_record_id: str = "usage_sample") -> UsageRecord: + return UsageRecord( + usage_record_id=usage_record_id, + created_at=100, + workflow_run_id=None, + request_channel="sync", + route_mode=None, + provider_name="openai", + model_name="gpt-x", + prompt_tokens=10, + completion_tokens=5, + total_tokens=15, + cost_amount=1.0, + currency_code="USD", + ) + + +def test_in_memory_telemetry_sink_trims_to_max_events() -> None: + # Covers cost_ledger.py:346 (event-list trimming when over capacity). + sink = InMemoryUsageTelemetrySink(max_events=2) + for index in range(3): + sink.emit_usage( + UsageTelemetryEvent(name=f"event-{index}", attributes={}, metrics={}) + ) + kept = [event.name for event in sink.events()] + assert kept == ["event-1", "event-2"] + + +def test_emit_usage_event_swallows_sink_failure() -> None: + # Covers cost_ledger.py:379-381 (best-effort telemetry export swallow). + ledger = _priced_ledger(telemetry_sink=_RaisingTelemetrySink()) + record = ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=10, completion_tokens=5 + ) + # The completion still succeeds even though the telemetry sink raised. + assert record.usage_record_id.startswith("usage_") + assert len(ledger.records()) == 1 + + +def test_non_blocking_store_rejects_zero_queue_size() -> None: + # Covers cost_ledger.py:395 (queue_size validation). + try: + NonBlockingLedgerStore(_RecordingBackend(), queue_size=0) + except ValueError as exc: + assert "queue_size must be at least 1" in str(exc) + else: # pragma: no cover + raise AssertionError("expected ValueError for queue_size < 1") + + +def test_non_blocking_store_drops_record_when_queue_full() -> None: + # Covers cost_ledger.py:412-422 (queue.Full drop path). + backend = _BlockingBackend() + sink = InMemoryUsageTelemetrySink() + store = NonBlockingLedgerStore(backend, queue_size=1, telemetry_sink=sink) + store.append(_sample_record("usage_a")) + # Worker has dequeued A and is now blocked inside append(A): the queue is empty. + assert backend.started.wait(timeout=5.0) + store.append(_sample_record("usage_b")) # fills the single queue slot + store.append(_sample_record("usage_c")) # overflows -> dropped + health = store.telemetry_health() + assert health["records_dropped"] == 1 + dropped_states = [ + event.attributes["contextual_orchestrator.usage.export_state"] + for event in sink.events() + ] + assert "dropped" in dropped_states + backend.release.set() + assert store.flush(timeout=5.0) + + +def test_non_blocking_store_query_delegates_to_backend() -> None: + # Covers cost_ledger.py:431 (query delegates to backend store). + backend = _RecordingBackend(rows=[{"created_at": 42}]) + store = NonBlockingLedgerStore(backend, queue_size=4) + assert store.query(start=0, end=100) == [{"created_at": 42}] + + +def test_non_blocking_store_flush_times_out_while_worker_busy() -> None: + # Covers cost_ledger.py:438 (flush returns False on timeout). + backend = _BlockingBackend() + store = NonBlockingLedgerStore(backend, queue_size=4) + store.append(_sample_record()) + assert backend.started.wait(timeout=5.0) # worker blocked; unfinished_tasks > 0 + assert store.flush(timeout=0.0) is False + backend.release.set() + assert store.flush(timeout=5.0) is True + + +def test_non_blocking_store_marks_stored_on_success() -> None: + # Covers cost_ledger.py:463-464 (worker success path marks stored + emits). + backend = _RecordingBackend() + sink = InMemoryUsageTelemetrySink() + store = NonBlockingLedgerStore(backend, queue_size=4, telemetry_sink=sink) + store.append(_sample_record()) + assert store.flush(timeout=5.0) + assert store.telemetry_health()["records_stored"] == 1 + stored_states = [ + event.attributes["contextual_orchestrator.usage.export_state"] + for event in sink.events() + ] + assert "stored" in stored_states + assert len(backend.appended) == 1 + + +def test_in_memory_ledger_store_len_reports_row_count() -> None: + # Covers cost_ledger.py:493 (InMemoryLedgerStore.__len__). Appends are made + # directly because an empty store is falsy (``store or ...`` in CostLedger). + store = InMemoryLedgerStore() + assert len(store) == 0 + store.append(_sample_record("usage_1")) + store.append(_sample_record("usage_2")) + assert len(store) == 2 + + +def test_sql_ledger_store_query_filters_by_time_window() -> None: + # Covers cost_ledger.py:617-618, 620-621 (start/end WHERE-clause builder). + conn = sqlite3.connect(":memory:") + store = SqlLedgerStore(conn, paramstyle="qmark") + ledger = _priced_ledger(store=store) + for created_at in (100, 200, 300): + ledger.record_usage( + provider="openai", + model="gpt-x", + prompt_tokens=1, + completion_tokens=1, + created_at=created_at, + ) + windowed = store.query(start=150, end=300) # half-open: only created_at == 200 + assert len(windowed) == 1 + assert windowed[0]["created_at"] == 200 + + +def test_non_blocking_store_wrapping_via_flag() -> None: + # Covers cost_ledger.py:659 (CostLedger wraps store when non_blocking_store=True). + ledger = _priced_ledger(non_blocking_store=True) + assert isinstance(ledger.store, NonBlockingLedgerStore) + ledger.record_usage(provider="openai", model="gpt-x", prompt_tokens=1000, completion_tokens=0) + assert ledger.flush(timeout=5.0) + assert ledger.total()["cost_amount"] == 2.0 + + +def test_record_usage_accepts_attribution_dimensions_instance() -> None: + # Covers cost_ledger.py:686 (attribution passed as AttributionDimensions). + ledger = _priced_ledger() + record = ledger.record_usage( + provider="openai", + model="gpt-x", + prompt_tokens=10, + completion_tokens=10, + attribution=AttributionDimensions(team="alpha", company="acme"), + ) + row = record.as_dict() + assert row["team_name"] == "alpha" + assert row["company_name"] == "acme" + + +def test_inline_store_failure_marks_health_and_emits_error() -> None: + # Covers cost_ledger.py:714-716 (inline append failure) and 837-839 + # (_mark_inline_failure counters). + sink = InMemoryUsageTelemetrySink() + ledger = _priced_ledger(store=_FailingLedgerStore(), telemetry_sink=sink) + record = ledger.record_usage( + provider="openai", model="gpt-x", prompt_tokens=10, completion_tokens=5 + ) + assert record.usage_record_id.startswith("usage_") + health = ledger.telemetry_health() + assert health["records_accepted"] == 1 + assert health["store_failures"] == 1 + assert health["last_error_type"] == "RuntimeError" + error_states = [ + event.attributes["contextual_orchestrator.usage.export_state"] + for event in sink.events() + ] + assert "export_error" in error_states + # The DB/client message must never leak into telemetry payload. + assert "secret prompt" not in repr(sink.events()) + + +def test_flush_returns_true_for_store_without_flush_method() -> None: + # Covers cost_ledger.py:738 (flush no-op when the store has no flush()). + ledger = _priced_ledger() # default InMemoryLedgerStore has no flush() + assert ledger.flush() is True + assert ledger.flush(timeout=1.0) is True + + if __name__ == "__main__": # pragma: no cover for _name, _fn in sorted(globals().items()): if _name.startswith("test_") and callable(_fn):