From 7b4f5af4f60802e2ee7a7e87c92d6af18977b2bb Mon Sep 17 00:00:00 2001 From: Michael Sitarzewski Date: Sun, 21 Jun 2026 19:22:03 -0500 Subject: [PATCH 1/3] feat(catalog): refresh model catalog to current frontier Add Claude Opus 4.8/4.7, GPT-5.5, GPT-5.4 mini, and Gemini 3.5 Flash; drop deprecated o3. Correct Anthropic GA context windows (Opus 4.6 and Sonnet 4.6 -> 1M; Sonnet 4.5 stays 200k, its 1M is beta-gated). Add gpt-5.5 to NO_TEMPERATURE_MODELS and _REASONING_EFFORT_MODELS after empirically confirming it rejects temperature. Model IDs sourced from live provider APIs, pricing cross-checked against the truefoundry/models feed, temperature behavior verified against the live OpenAI endpoint. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EkrekgzMAQko92UkjnXhHL --- src/duh/providers/catalog.py | 55 +++++++++++++++++++++++------ src/duh/providers/openai.py | 1 + tests/unit/test_providers_google.py | 3 +- 3 files changed, 47 insertions(+), 12 deletions(-) diff --git a/src/duh/providers/catalog.py b/src/duh/providers/catalog.py index 93927b9..2bdcedb 100644 --- a/src/duh/providers/catalog.py +++ b/src/duh/providers/catalog.py @@ -53,10 +53,26 @@ MODEL_CATALOG: dict[str, list[dict[str, Any]]] = { "anthropic": [ + { + "model_id": "claude-opus-4-8", + "display_name": "Claude Opus 4.8", + "context_window": 1_000_000, + "max_output_tokens": 128_000, + "input_cost_per_mtok": 5.00, + "output_cost_per_mtok": 25.00, + }, + { + "model_id": "claude-opus-4-7", + "display_name": "Claude Opus 4.7", + "context_window": 1_000_000, + "max_output_tokens": 128_000, + "input_cost_per_mtok": 5.00, + "output_cost_per_mtok": 25.00, + }, { "model_id": "claude-opus-4-6", "display_name": "Claude Opus 4.6", - "context_window": 200_000, + "context_window": 1_000_000, "max_output_tokens": 128_000, "input_cost_per_mtok": 5.00, "output_cost_per_mtok": 25.00, @@ -64,7 +80,7 @@ { "model_id": "claude-sonnet-4-6", "display_name": "Claude Sonnet 4.6", - "context_window": 200_000, + "context_window": 1_000_000, "max_output_tokens": 64_000, "input_cost_per_mtok": 3.00, "output_cost_per_mtok": 15.00, @@ -87,14 +103,30 @@ }, ], "openai": [ + { + "model_id": "gpt-5.5", + "display_name": "GPT-5.5", + "context_window": 1_050_000, + "max_output_tokens": 128_000, + "input_cost_per_mtok": 5.00, + "output_cost_per_mtok": 30.00, + }, { "model_id": "gpt-5.4", "display_name": "GPT-5.4", - "context_window": 1_048_576, + "context_window": 1_050_000, "max_output_tokens": 128_000, "input_cost_per_mtok": 2.50, "output_cost_per_mtok": 15.00, }, + { + "model_id": "gpt-5.4-mini", + "display_name": "GPT-5.4 mini", + "context_window": 400_000, + "max_output_tokens": 128_000, + "input_cost_per_mtok": 0.75, + "output_cost_per_mtok": 4.50, + }, { "model_id": "gpt-5.2", "display_name": "GPT-5.2", @@ -111,16 +143,16 @@ "input_cost_per_mtok": 0.25, "output_cost_per_mtok": 2.00, }, - { - "model_id": "o3", - "display_name": "o3", - "context_window": 200_000, - "max_output_tokens": 100_000, - "input_cost_per_mtok": 2.00, - "output_cost_per_mtok": 8.00, - }, ], "google": [ + { + "model_id": "gemini-3.5-flash", + "display_name": "Gemini 3.5 Flash", + "context_window": 1_048_576, + "max_output_tokens": 65_536, + "input_cost_per_mtok": 1.50, + "output_cost_per_mtok": 9.00, + }, { "model_id": "gemini-3.1-pro-preview", "display_name": "Gemini 3.1 Pro (Preview)", @@ -243,6 +275,7 @@ "gpt-5-nano", "gpt-5.2", "gpt-5.4", + "gpt-5.5", } # ── Providers that are challenger-only (not proposer-eligible) ── diff --git a/src/duh/providers/openai.py b/src/duh/providers/openai.py index 9df11e4..7675629 100644 --- a/src/duh/providers/openai.py +++ b/src/duh/providers/openai.py @@ -51,6 +51,7 @@ "gpt-5-nano", "gpt-5.2", "gpt-5.4", + "gpt-5.5", } diff --git a/tests/unit/test_providers_google.py b/tests/unit/test_providers_google.py index 9c31b04..b430cc5 100644 --- a/tests/unit/test_providers_google.py +++ b/tests/unit/test_providers_google.py @@ -83,8 +83,9 @@ def test_provider_id(): async def test_list_models(): prov = GoogleProvider(client=_make_client()) models = await prov.list_models() - assert len(models) == 5 + assert len(models) == 6 ids = {m.model_id for m in models} + assert "gemini-3.5-flash" in ids assert "gemini-3.1-pro-preview" in ids assert "gemini-3-pro-preview" in ids assert "gemini-3-flash-preview" in ids From c785c7ec2c6fd6add0565f9025cabe73fb77d613 Mon Sep 17 00:00:00 2001 From: Michael Sitarzewski Date: Sun, 21 Jun 2026 19:22:03 -0500 Subject: [PATCH 2/3] feat(persist): persist consensus incrementally instead of at completion Introduce IncrementalPersister (memory/persist.py): the thread is created up front as "active", each round is committed as soon as it finishes, and the thread is finalized to "complete" at the end. A crash mid-run now leaves a real partial thread instead of nothing. The WebSocket streams a thread_started event with the real thread ID so clients can deep-link mid-run. Consolidates the three previously-duplicated persist paths (CLI, WS, and the batch convenience) into one shared module and adds ConsensusContext.snapshot_round() so a finished round can be persisted before it is archived. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EkrekgzMAQko92UkjnXhHL --- src/duh/api/routes/ws.py | 130 +++++++----------------------- src/duh/cli/app.py | 141 +++++++++++---------------------- src/duh/consensus/machine.py | 36 +++++---- src/duh/memory/persist.py | 149 +++++++++++++++++++++++++++++++++++ tests/unit/test_persist.py | 141 +++++++++++++++++++++++++++++++++ 5 files changed, 386 insertions(+), 211 deletions(-) create mode 100644 src/duh/memory/persist.py create mode 100644 tests/unit/test_persist.py diff --git a/src/duh/api/routes/ws.py b/src/duh/api/routes/ws.py index c43c4af..3d9fbde 100644 --- a/src/duh/api/routes/ws.py +++ b/src/duh/api/routes/ws.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json import logging from typing import TYPE_CHECKING, Any @@ -10,7 +9,7 @@ if TYPE_CHECKING: from duh.config.schema import DuhConfig - from duh.consensus.machine import RoundResult + from duh.memory.persist import IncrementalPersister from duh.providers.manager import ProviderManager from duh.tools.registry import ToolRegistry @@ -126,6 +125,22 @@ async def _stream_consensus( effective_panel = panel or config.consensus.panel or None use_native_search = config.tools.enabled and config.tools.web_search.native + # Persist incrementally so a mid-run disconnect/crash leaves a real + # partial thread. Create it up front and tell the client its ID early. + persister: IncrementalPersister | None = None + db_factory = getattr(ws.app.state, "db_factory", None) + if db_factory is not None: + from duh.memory.persist import IncrementalPersister + + persister = IncrementalPersister(db_factory, question) + try: + thread_id_early = await persister.start() + ctx.thread_id = thread_id_early + await ws.send_json({"type": "thread_started", "thread_id": thread_id_early}) + except Exception: + logger.exception("Failed to create consensus thread") + persister = None + for _round in range(config.general.max_rounds): # PROPOSE sm.transition(ConsensusState.PROPOSE) @@ -205,6 +220,11 @@ async def _stream_consensus( # COMMIT sm.transition(ConsensusState.COMMIT) await handle_commit(ctx, pm) + if persister is not None: + try: + await persister.persist_round(ctx.snapshot_round()) + except Exception: + logger.exception("Failed to persist consensus round") await ws.send_json( { "type": "commit", @@ -226,16 +246,13 @@ async def _stream_consensus( await generate_followups(ctx, pm) - # Persist to DB if available - thread_id: str | None = None - db_factory = getattr(ws.app.state, "db_factory", None) - if db_factory is not None: + # Finalize the incrementally-persisted thread (mark complete, attach + # overview / follow-ups / usage). The thread + all rounds are already saved. + thread_id: str | None = ctx.thread_id or None + if persister is not None: try: - thread_id = await _persist_consensus( - db_factory, - question, - ctx.round_history, - ctx.overview, + await persister.finalize( + overview=ctx.overview, followups=ctx.followups or None, usage={ "input_tokens": pm.total_input_tokens, @@ -244,7 +261,7 @@ async def _stream_consensus( }, ) except Exception: - logger.exception("Failed to persist consensus thread") + logger.exception("Failed to finalize consensus thread") await ws.send_json( { @@ -357,92 +374,3 @@ async def _run(idx: int, ref: str) -> tuple[int, tuple[str, str, Any]]: raise ConsensusError(msg) ctx.challenges = challenges # type: ignore[attr-defined] - - -async def _persist_consensus( - db_factory: object, - question: str, - round_history: list[RoundResult], - overview: str | None = None, - followups: list[str] | None = None, - usage: dict[str, float] | None = None, -) -> str: - """Persist consensus round history to the database. - - Returns the new thread ID. - """ - from duh.memory.repository import MemoryRepository - - async with db_factory() as session: # type: ignore[operator] - repo = MemoryRepository(session) - thread = await repo.create_thread(question) - thread.status = "complete" - - for rr in round_history: - turn = await repo.create_turn(thread.id, rr.round_number, "COMMIT") - proposal_cit = None - if rr.proposal_citations: - proposal_cit = json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in rr.proposal_citations - ] - ) - await repo.add_contribution( - turn.id, - rr.proposal_model, - "proposer", - rr.proposal, - citations_json=proposal_cit, - ) - for ch in rr.challenges: - ch_cit = None - if ch.citations: - ch_cit = json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in ch.citations - ] - ) - await repo.add_contribution( - turn.id, - ch.model_ref, - "challenger", - ch.content, - citations_json=ch_cit, - ) - rev_cit = None - if rr.revision_citations: - rev_cit = json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in rr.revision_citations - ] - ) - await repo.add_contribution( - turn.id, - rr.proposal_model, - "reviser", - rr.revision, - citations_json=rev_cit, - ) - await repo.save_decision( - turn.id, - thread.id, - rr.decision, - rr.confidence, - rigor=rr.rigor, - dissent=rr.dissent, - ) - - if overview: - await repo.save_thread_summary(thread.id, overview, "overview") - - if followups: - thread.followups_json = json.dumps(followups) - - if usage: - thread.usage_json = json.dumps(usage) - - await session.commit() - return str(thread.id) diff --git a/src/duh/cli/app.py b/src/duh/cli/app.py index 2f61a8e..1f9fce7 100644 --- a/src/duh/cli/app.py +++ b/src/duh/cli/app.py @@ -27,6 +27,7 @@ from duh.config.schema import DuhConfig from duh.consensus.machine import RoundResult from duh.memory.models import Thread, Vote + from duh.memory.persist import IncrementalPersister from duh.providers.base import ModelInfo from duh.providers.manager import ProviderManager from duh.tools.registry import ToolRegistry @@ -229,98 +230,24 @@ async def persist_consensus( followups: list[str] | None = None, usage: dict[str, float] | None = None, ) -> str: - """Persist full consensus round history to the database. + """Persist a full consensus round history to the database. - Saves proposals, challenger responses, revisions, citations, - decisions, and overview — the same rich format used by the web UI. + Thin wrapper over :func:`duh.memory.persist.persist_consensus` (the + canonical incremental implementation). Retained for the batch command and + backward-compatible imports. Returns the new thread ID. """ - import json as _json - - from duh.memory.repository import MemoryRepository - - async with db_factory() as session: - repo = MemoryRepository(session) - thread = await repo.create_thread(question) - thread.status = "complete" - - for rr in round_history: - turn = await repo.create_turn(thread.id, rr.round_number, "COMMIT") - - # Proposal with citations - proposal_cit = None - if rr.proposal_citations: - proposal_cit = _json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in rr.proposal_citations - ] - ) - await repo.add_contribution( - turn.id, - rr.proposal_model, - "proposer", - rr.proposal, - citations_json=proposal_cit, - ) - - # Challenger responses with citations - for ch in rr.challenges: - ch_cit = None - if ch.citations: - ch_cit = _json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in ch.citations - ] - ) - await repo.add_contribution( - turn.id, - ch.model_ref, - "challenger", - ch.content, - citations_json=ch_cit, - ) - - # Revision with citations - rev_cit = None - if rr.revision_citations: - rev_cit = _json.dumps( - [ - {"url": c["url"], "title": c.get("title")} - for c in rr.revision_citations - ] - ) - await repo.add_contribution( - turn.id, - rr.proposal_model, - "reviser", - rr.revision, - citations_json=rev_cit, - ) - - # Decision - await repo.save_decision( - turn.id, - thread.id, - rr.decision, - rr.confidence, - rigor=rr.rigor, - dissent=rr.dissent, - ) - - if overview: - await repo.save_thread_summary(thread.id, overview, "overview") - - if followups: - thread.followups_json = _json.dumps(followups) - - if usage: - thread.usage_json = _json.dumps(usage) - - await session.commit() - return str(thread.id) + from duh.memory.persist import persist_consensus as _persist + + return await _persist( + db_factory, + question, + round_history, + overview=overview, + followups=followups, + usage=usage, + ) async def _run_consensus( @@ -377,6 +304,21 @@ async def _run_consensus( # Resolve effective panel from config or explicit arg effective_panel = panel or config.consensus.panel or None + # Persist incrementally so a mid-run crash leaves a real partial thread + # instead of nothing. The thread is created up front (status "active"). + persister: IncrementalPersister | None = None + if db_factory is not None: + from duh.memory.persist import IncrementalPersister + + persister = IncrementalPersister(db_factory, question) + try: + ctx.thread_id = await persister.start() + except Exception: + import logging as _logging + + _logging.getLogger(__name__).exception("Failed to create consensus thread") + persister = None + for _round in range(config.general.max_rounds): # PROPOSE sm.transition(ConsensusState.PROPOSE) @@ -451,6 +393,15 @@ async def _run_consensus( # COMMIT sm.transition(ConsensusState.COMMIT) await handle_commit(ctx, pm) + if persister is not None: + try: + await persister.persist_round(ctx.snapshot_round()) + except Exception: + import logging as _logging + + _logging.getLogger(__name__).exception( + "Failed to persist consensus round" + ) if display: display.show_commit(ctx.confidence, ctx.rigor, ctx.dissent) display.round_footer( @@ -494,13 +445,11 @@ async def _run_consensus( all_citations.extend(ch.citations) all_citations.extend(ctx.revision_citations) - # Persist full round history if DB available - if db_factory is not None: + # Finalize the incrementally-persisted thread: mark complete and attach + # the overview, follow-ups, and usage totals. + if persister is not None: try: - await persist_consensus( - db_factory, - question, - ctx.round_history, + await persister.finalize( overview=ctx.overview, followups=ctx.followups or None, usage={ @@ -512,7 +461,9 @@ async def _run_consensus( except Exception: import logging as _logging - _logging.getLogger(__name__).exception("Failed to persist consensus thread") + _logging.getLogger(__name__).exception( + "Failed to finalize consensus thread" + ) return ( ctx.decision or "", diff --git a/src/duh/consensus/machine.py b/src/duh/consensus/machine.py index f0c6ae3..c74d785 100644 --- a/src/duh/consensus/machine.py +++ b/src/duh/consensus/machine.py @@ -132,23 +132,29 @@ def _clear_round_data(self) -> None: self.dissent = None self.converged = False + def snapshot_round(self) -> RoundResult: + """Build a RoundResult from the current (in-progress) round data. + + Used both to archive a finished round to ``round_history`` and to + persist a round incrementally before it is archived. + """ + return RoundResult( + round_number=self.current_round, + proposal=self.proposal or "", + proposal_model=self.proposal_model or "", + challenges=tuple(self.challenges), + revision=self.revision or "", + decision=self.decision or "", + confidence=self.confidence, + rigor=self.rigor, + dissent=self.dissent, + proposal_citations=tuple(self.proposal_citations), + revision_citations=tuple(self.revision_citations), + ) + def _archive_round(self) -> None: """Archive current round data to history.""" - self.round_history.append( - RoundResult( - round_number=self.current_round, - proposal=self.proposal or "", - proposal_model=self.proposal_model or "", - challenges=tuple(self.challenges), - revision=self.revision or "", - decision=self.decision or "", - confidence=self.confidence, - rigor=self.rigor, - dissent=self.dissent, - proposal_citations=tuple(self.proposal_citations), - revision_citations=tuple(self.revision_citations), - ) - ) + self.round_history.append(self.snapshot_round()) # ── State machine ───────────────────────────────────────────── diff --git a/src/duh/memory/persist.py b/src/duh/memory/persist.py new file mode 100644 index 0000000..383afd5 --- /dev/null +++ b/src/duh/memory/persist.py @@ -0,0 +1,149 @@ +"""Incremental consensus persistence. + +Writes a consensus thread to the database progressively — the thread is +created up front (status ``active``), each round is committed as soon as it +finishes, and the thread is finalized (status ``complete``) at the end. A +crash mid-run therefore leaves a real, partial thread instead of nothing. + +The per-round write logic lives in :meth:`IncrementalPersister.persist_round` +so every caller (CLI loop, WebSocket loop, and the batch +:func:`persist_consensus` convenience) shares one implementation. +""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING + +from duh.memory.repository import MemoryRepository + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + + from duh.consensus.machine import RoundResult + +logger = logging.getLogger(__name__) + + +def _citations_json(citations: object) -> str | None: + """Serialize a list of citation dicts to JSON, or None if empty.""" + if not citations: + return None + return json.dumps( + [{"url": c["url"], "title": c.get("title")} for c in citations] # type: ignore[attr-defined] + ) + + +class IncrementalPersister: + """Persist a consensus thread one round at a time. + + Each method opens its own session and commits independently, so a round + that has been persisted survives a later crash. + """ + + def __init__( + self, db_factory: async_sessionmaker[AsyncSession], question: str + ) -> None: + self._db_factory = db_factory + self._question = question + self.thread_id: str | None = None + + async def start(self) -> str: + """Create the thread (status ``active``) and return its ID.""" + async with self._db_factory() as session: + repo = MemoryRepository(session) + thread = await repo.create_thread(self._question) + thread.status = "active" + await session.commit() + self.thread_id = str(thread.id) + return self.thread_id + + async def persist_round(self, rr: RoundResult) -> None: + """Write a single finished round (turn + contributions + decision).""" + if self.thread_id is None: + msg = "persist_round called before start()" + raise RuntimeError(msg) + async with self._db_factory() as session: + repo = MemoryRepository(session) + turn = await repo.create_turn(self.thread_id, rr.round_number, "COMMIT") + await repo.add_contribution( + turn.id, + rr.proposal_model, + "proposer", + rr.proposal, + citations_json=_citations_json(rr.proposal_citations), + ) + for ch in rr.challenges: + await repo.add_contribution( + turn.id, + ch.model_ref, + "challenger", + ch.content, + citations_json=_citations_json(ch.citations), + ) + await repo.add_contribution( + turn.id, + rr.proposal_model, + "reviser", + rr.revision, + citations_json=_citations_json(rr.revision_citations), + ) + await repo.save_decision( + turn.id, + self.thread_id, + rr.decision, + rr.confidence, + rigor=rr.rigor, + dissent=rr.dissent, + ) + await session.commit() + + async def finalize( + self, + *, + overview: str | None = None, + followups: list[str] | None = None, + usage: dict[str, float] | None = None, + ) -> None: + """Mark the thread complete and attach overview/followups/usage.""" + if self.thread_id is None: + msg = "finalize called before start()" + raise RuntimeError(msg) + async with self._db_factory() as session: + repo = MemoryRepository(session) + thread = await repo.get_thread(self.thread_id) + if thread is None: + logger.warning("finalize: thread %s vanished", self.thread_id) + return + thread.status = "complete" + if overview: + await repo.save_thread_summary(thread.id, overview, "overview") + if followups: + thread.followups_json = json.dumps(followups) + if usage: + thread.usage_json = json.dumps(usage) + await session.commit() + + +async def persist_consensus( + db_factory: async_sessionmaker[AsyncSession], + question: str, + round_history: list[RoundResult], + *, + overview: str | None = None, + followups: list[str] | None = None, + usage: dict[str, float] | None = None, +) -> str: + """Persist a full round history at once and return the thread ID. + + Convenience wrapper over :class:`IncrementalPersister` for callers that + already hold the complete history (batch runs, tests). Internally this is + still the same incremental path: start -> persist each round -> finalize. + """ + persister = IncrementalPersister(db_factory, question) + thread_id = await persister.start() + for rr in round_history: + await persister.persist_round(rr) + await persister.finalize(overview=overview, followups=followups, usage=usage) + return thread_id diff --git a/tests/unit/test_persist.py b/tests/unit/test_persist.py new file mode 100644 index 0000000..7842631 --- /dev/null +++ b/tests/unit/test_persist.py @@ -0,0 +1,141 @@ +"""Tests for incremental consensus persistence (duh.memory.persist).""" + +from __future__ import annotations + +from sqlalchemy import event +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from duh.consensus.machine import ChallengeResult, RoundResult +from duh.memory.models import Base +from duh.memory.persist import IncrementalPersister, persist_consensus +from duh.memory.repository import MemoryRepository + + +async def _make_factory() -> async_sessionmaker: + engine = create_async_engine("sqlite+aiosqlite://") + + @event.listens_for(engine.sync_engine, "connect") + def _fks(dbapi_conn, _record): # type: ignore[no-untyped-def] + cur = dbapi_conn.cursor() + cur.execute("PRAGMA foreign_keys=ON") + cur.close() + + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return async_sessionmaker(engine, expire_on_commit=False) + + +def _round(n: int = 1) -> RoundResult: + return RoundResult( + round_number=n, + proposal=f"Proposal {n}", + proposal_model="anthropic:claude-opus-4-8", + challenges=( + ChallengeResult( + model_ref="openai:gpt-5.5", + content=f"Challenge {n}", + citations=({"url": "https://ex.com", "title": "Ex"},), + ), + ), + revision=f"Revision {n}", + decision=f"Decision {n}", + confidence=0.8, + rigor=0.7, + dissent="Some dissent", + proposal_citations=({"url": "https://p.com", "title": "P"},), + revision_citations=(), + ) + + +class TestIncrementalPersister: + async def test_start_creates_active_thread(self) -> None: + factory = await _make_factory() + p = IncrementalPersister(factory, "Q?") + tid = await p.start() + assert tid + async with factory() as session: + thread = await MemoryRepository(session).get_thread(tid) + assert thread is not None + assert thread.status == "active" + + async def test_round_visible_before_finalize(self) -> None: + """A persisted round is durable even if finalize never runs (crash).""" + factory = await _make_factory() + p = IncrementalPersister(factory, "Q?") + tid = await p.start() + await p.persist_round(_round(1)) + # Simulate a crash here — no finalize. + async with factory() as session: + thread = await MemoryRepository(session).get_thread(tid) + assert thread is not None + assert thread.status == "active" # not yet complete + assert len(thread.turns) == 1 + turn = thread.turns[0] + roles = sorted(c.role for c in turn.contributions) + assert roles == ["challenger", "proposer", "reviser"] + assert turn.decision is not None + assert turn.decision.content == "Decision 1" + # Citations persisted on the proposer contribution + prop = next(c for c in turn.contributions if c.role == "proposer") + assert prop.citations_json is not None + assert "p.com" in prop.citations_json + + async def test_finalize_completes_and_attaches_usage(self) -> None: + factory = await _make_factory() + p = IncrementalPersister(factory, "Q?") + tid = await p.start() + await p.persist_round(_round(1)) + await p.finalize( + overview="Overview text", + followups=["next?"], + usage={"input_tokens": 100, "output_tokens": 50, "cost_usd": 0.01}, + ) + async with factory() as session: + thread = await MemoryRepository(session).get_thread(tid) + assert thread is not None + assert thread.status == "complete" + assert thread.usage_json is not None + assert "100" in thread.usage_json + assert thread.followups_json is not None + assert "next?" in thread.followups_json + + async def test_persist_round_before_start_raises(self) -> None: + factory = await _make_factory() + p = IncrementalPersister(factory, "Q?") + try: + await p.persist_round(_round(1)) + raise AssertionError("expected RuntimeError") + except RuntimeError: + pass + + async def test_multiple_rounds_accumulate(self) -> None: + factory = await _make_factory() + p = IncrementalPersister(factory, "Q?") + tid = await p.start() + await p.persist_round(_round(1)) + await p.persist_round(_round(2)) + await p.finalize() + async with factory() as session: + thread = await MemoryRepository(session).get_thread(tid) + assert thread is not None + assert len(thread.turns) == 2 + assert {t.round_number for t in thread.turns} == {1, 2} + + +class TestPersistConsensusConvenience: + async def test_batch_wrapper_matches_incremental(self) -> None: + factory = await _make_factory() + tid = await persist_consensus( + factory, + "Q?", + [_round(1), _round(2)], + overview="ov", + followups=["f1"], + usage={"input_tokens": 9, "output_tokens": 3, "cost_usd": 0.001}, + ) + async with factory() as session: + thread = await MemoryRepository(session).get_thread(tid) + assert thread is not None + assert thread.status == "complete" + assert len(thread.turns) == 2 + assert thread.usage_json is not None and "9" in thread.usage_json From dfc5227fb2130616fd17b0469b6a01feeb47e371 Mon Sep 17 00:00:00 2001 From: Michael Sitarzewski Date: Sun, 21 Jun 2026 19:22:03 -0500 Subject: [PATCH 3/3] feat(scripts): add manual catalog refresh tool scripts/refresh_catalog.py diffs the model catalog against the truefoundry/models feed and the live provider APIs. It reports price/context/status changes, hashes a per-model field projection against catalog_snapshot.json to show what changed since the last run, discovers new frontier models via the live APIs, and empirically probes OpenAI temperature support (the one field the feed gets wrong). Propose-only: it never modifies catalog.py. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EkrekgzMAQko92UkjnXhHL --- scripts/catalog_snapshot.json | 27 +++ scripts/refresh_catalog.py | 305 ++++++++++++++++++++++++++++++++++ 2 files changed, 332 insertions(+) create mode 100644 scripts/catalog_snapshot.json create mode 100644 scripts/refresh_catalog.py diff --git a/scripts/catalog_snapshot.json b/scripts/catalog_snapshot.json new file mode 100644 index 0000000..51995f3 --- /dev/null +++ b/scripts/catalog_snapshot.json @@ -0,0 +1,27 @@ +{ + "anthropic:claude-haiku-4-5-20251001": "982098c920605ca8", + "anthropic:claude-opus-4-6": "93f65f9d16d5bd0f", + "anthropic:claude-opus-4-7": "93f65f9d16d5bd0f", + "anthropic:claude-opus-4-8": "93f65f9d16d5bd0f", + "anthropic:claude-sonnet-4-5-20250929": "d3c27f27755bd8ad", + "anthropic:claude-sonnet-4-6": "559b20307ff0dc73", + "google:gemini-2.5-flash": "e8eee2f7539ad41c", + "google:gemini-2.5-pro": "72013ec726238872", + "google:gemini-3-flash-preview": "6d59037ac049bc5e", + "google:gemini-3-pro-preview": "496276d00d83b9eb", + "google:gemini-3.1-pro-preview": "ec99bba22258dee5", + "google:gemini-3.5-flash": "41689dfcb2602437", + "mistral:codestral-latest": "no-feed", + "mistral:mistral-large-latest": "no-feed", + "mistral:mistral-medium-latest": "no-feed", + "mistral:mistral-small-latest": "no-feed", + "openai:gpt-5-mini": "8757c36c175ab8b7", + "openai:gpt-5.2": "82076fbd9d17620b", + "openai:gpt-5.4": "0afb2bc2c95193d9", + "openai:gpt-5.4-mini": "746ba23091cff0bd", + "openai:gpt-5.5": "b1213071b7885e5f", + "perplexity:sonar": "no-feed", + "perplexity:sonar-deep-research": "no-feed", + "perplexity:sonar-pro": "no-feed", + "perplexity:sonar-reasoning-pro": "no-feed" +} diff --git a/scripts/refresh_catalog.py b/scripts/refresh_catalog.py new file mode 100644 index 0000000..1c3bdd5 --- /dev/null +++ b/scripts/refresh_catalog.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +"""Manual model-catalog refresh tool — propose-only, never auto-applies. + +What it does (run on demand, no cron): + +1. Reads the models currently in ``duh.providers.catalog.MODEL_CATALOG``. +2. For each, pulls a *projection* (price / context / max-output / status) from + the community truefoundry/models feed and diffs it against our catalog. +3. Hashes that projection and compares to ``scripts/catalog_snapshot.json`` so + you can see what changed since the last run. +4. Queries the live provider APIs to surface NEW frontier models (in the API, + not in our catalog) and models we list that the API no longer returns. +5. Empirically probes OpenAI temperature support against the real endpoint — + because the feed is unreliable on *behavioral* fields (it had gpt-5.5 wrong) + — and flags any mismatch with ``NO_TEMPERATURE_MODELS``. + +It prints a report and (with --write-snapshot) updates the snapshot. It does +NOT modify ``catalog.py``. Pricing/context/status are DATA: diffed for you to +confirm. Temperature is BEHAVIOR: verified live before trusting. + +Usage: + uv run python scripts/refresh_catalog.py + uv run python scripts/refresh_catalog.py --no-probe # offline-ish + uv run python scripts/refresh_catalog.py --write-snapshot # save baseline +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv + +from duh.providers.catalog import MODEL_CATALOG, NO_TEMPERATURE_MODELS + +# truefoundry provider directory names (differ from our provider_id keys) +_TF_BASE = "https://raw.githubusercontent.com/truefoundry/models/main/providers" +_TF_DIR = { + "anthropic": "anthropic", + "openai": "openai", + "google": "google-gemini", + "mistral": "mistral", + "perplexity": "perplexity", +} +_SNAPSHOT = Path(__file__).parent / "catalog_snapshot.json" +_PRICE_TOL = 1e-4 # $/Mtok difference worth reporting + + +# ── HTTP helpers (stdlib only) ───────────────────────────────── + + +def _get_text(url: str, headers: dict[str, str] | None = None) -> str | None: + req = urllib.request.Request(url, headers=headers or {}) + try: + with urllib.request.urlopen(req, timeout=25) as resp: + return resp.read().decode() + except (urllib.error.URLError, TimeoutError): + return None + + +def _get_json(url: str, headers: dict[str, str] | None = None) -> Any | None: + text = _get_text(url, headers) + if text is None: + return None + try: + return json.loads(text) + except json.JSONDecodeError: + return None + + +def _post_json(url: str, headers: dict[str, str], body: dict[str, Any]) -> Any | None: + data = json.dumps(body).encode() + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=25) as resp: + return json.loads(resp.read().decode()) + except urllib.error.HTTPError as e: + try: + return json.loads(e.read().decode()) + except (json.JSONDecodeError, ValueError): + return None + except (urllib.error.URLError, TimeoutError): + return None + + +# ── truefoundry feed ─────────────────────────────────────────── + + +def _num(text: str, key: str) -> float | None: + m = re.search(rf"^\s*{re.escape(key)}:\s*([0-9.eE+-]+)", text, re.M) + return float(m.group(1)) if m else None + + +def _str(text: str, key: str) -> str | None: + m = re.search(rf'^\s*{re.escape(key)}:\s*"?(\w[\w.-]*)"?', text, re.M) + return m.group(1) if m else None + + +def tf_projection(provider: str, model_id: str) -> dict[str, Any] | None: + """Fetch {input_mtok, output_mtok, context, max_output, status} from feed.""" + tf_dir = _TF_DIR.get(provider) + if tf_dir is None: + return None + text = _get_text(f"{_TF_BASE}/{tf_dir}/{model_id}.yaml") + if text is None: + return None + ic = _num(text, "input_cost_per_token") + oc = _num(text, "output_cost_per_token") + return { + "input_cost_per_mtok": round(ic * 1e6, 4) if ic is not None else None, + "output_cost_per_mtok": round(oc * 1e6, 4) if oc is not None else None, + "context_window": int(_num(text, "context_window") or 0) or None, + "max_output_tokens": int(_num(text, "max_output_tokens") or 0) or None, + "status": _str(text, "status"), + } + + +def _hash(projection: dict[str, Any]) -> str: + payload = json.dumps(projection, sort_keys=True) + return hashlib.sha256(payload.encode()).hexdigest()[:16] + + +# ── live provider model lists ────────────────────────────────── + + +def live_ids(provider: str) -> set[str] | None: + """Return the set of model IDs the live API reports, or None if no key.""" + if provider == "anthropic": + key = os.environ.get("ANTHROPIC_API_KEY") + if not key: + return None + data = _get_json( + "https://api.anthropic.com/v1/models?limit=100", + {"x-api-key": key, "anthropic-version": "2023-06-01"}, + ) + return {m["id"] for m in data["data"]} if data else None + if provider == "openai": + key = os.environ.get("OPENAI_API_KEY") + if not key: + return None + data = _get_json( + "https://api.openai.com/v1/models", {"Authorization": f"Bearer {key}"} + ) + if not data: + return None + # Frontier chat/reasoning families only — skip legacy & non-chat variants + keep = re.compile(r"^(gpt-5(\.\d+)?|o[34])(-mini|-nano|-pro)?$") + return {m["id"] for m in data["data"] if keep.match(m["id"])} + if provider == "google": + key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY") + if not key: + return None + data = _get_json( + f"https://generativelanguage.googleapis.com/v1beta/models?key={key}&pageSize=200" + ) + if not data: + return None + skip = re.compile(r"tts|image|embedding|aqa|learnlm|gemma|robotics|computer") + return { + m["name"].removeprefix("models/") + for m in data.get("models", []) + if "generateContent" in m.get("supportedGenerationMethods", []) + and "gemini" in m["name"] + and not skip.search(m["name"]) + } + return None + + +def probe_openai_temperature(model_id: str) -> bool | None: + """True if the model accepts temperature!=1, False if it rejects, None on error.""" + key = os.environ.get("OPENAI_API_KEY") + if not key: + return None + resp = _post_json( + "https://api.openai.com/v1/chat/completions", + {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, + { + "model": model_id, + "messages": [{"role": "user", "content": "hi"}], + "temperature": 0.7, + "max_completion_tokens": 16, + }, + ) + if resp is None: + return None + err = resp.get("error") if isinstance(resp, dict) else None + if err: + return False if "temperature" in err.get("message", "") else None + return True + + +# ── report ───────────────────────────────────────────────────── + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--no-probe", action="store_true", help="skip live behavioral probes" + ) + parser.add_argument( + "--write-snapshot", action="store_true", help="update catalog_snapshot.json" + ) + args = parser.parse_args() + + load_dotenv() + snapshot: dict[str, str] = {} + if _SNAPSHOT.exists(): + snapshot = json.loads(_SNAPSHOT.read_text()) + new_snapshot: dict[str, str] = {} + + print("=" * 70) + print("CATALOG REFRESH — proposal only, nothing is modified") + print("=" * 70) + + data_diffs: list[str] = [] + changed_since_last: list[str] = [] + + for provider, models in MODEL_CATALOG.items(): + for m in models: + mid = provider + ":" + m["model_id"] + proj = tf_projection(provider, m["model_id"]) + if proj is None: + new_snapshot[mid] = "no-feed" + continue + h = _hash(proj) + new_snapshot[mid] = h + if snapshot.get(mid) not in (None, h): + changed_since_last.append(mid) + for field in ("input_cost_per_mtok", "output_cost_per_mtok"): + up = proj[field] + cur = m.get(field) + if up is not None and abs(up - cur) > _PRICE_TOL: + data_diffs.append(f" {mid} {field}: {cur} -> {up} (feed)") + for field in ("context_window", "max_output_tokens"): + up = proj[field] + cur = m.get(field) + if up is not None and up != cur: + data_diffs.append(f" {mid} {field}: {cur} -> {up} (feed)") + if proj["status"] == "deprecated": + data_diffs.append(f" {mid} STATUS=deprecated (drop candidate)") + + print("\n## DATA changes vs catalog (price / context / status) — confirm manually") + print("\n".join(data_diffs) if data_diffs else " (none — catalog matches feed)") + + print("\n## Changed since last snapshot") + print( + "\n".join(" " + m for m in changed_since_last) + if changed_since_last + else " (none)" + ) + + print("\n## DISCOVERY — live API vs catalog") + for provider in MODEL_CATALOG: + ids = live_ids(provider) + if ids is None: + print(f" {provider}: (no key / no list endpoint — skipped)") + continue + catalog_ids = {m["model_id"] for m in MODEL_CATALOG[provider]} + new = sorted(ids - catalog_ids) + gone = sorted(catalog_ids - ids) + if new: + print(f" {provider}: NEW candidates -> {', '.join(new)}") + if gone: + print(f" {provider}: in catalog but not in API -> {', '.join(gone)}") + if not new and not gone: + print(f" {provider}: in sync") + + if not args.no_probe: + print("\n## BEHAVIOR — OpenAI temperature probe vs NO_TEMPERATURE_MODELS") + openai_ids = {m["model_id"] for m in MODEL_CATALOG.get("openai", [])} + for mid in sorted(openai_ids): + accepts = probe_openai_temperature(mid) + if accepts is None: + print(f" {mid}: probe inconclusive") + continue + listed = mid in NO_TEMPERATURE_MODELS + rejects = not accepts + if rejects and not listed: + print(f" {mid}: REJECTS temp but NOT in NO_TEMPERATURE_MODELS — ADD") + elif accepts and listed: + print( + f" {mid}: accepts temp but IS in NO_TEMPERATURE_MODELS" + " — review (may be intentional w/ reasoning_effort)" + ) + else: + print(f" {mid}: ok ({'no-temp' if rejects else 'temp'})") + + if args.write_snapshot: + _SNAPSHOT.write_text(json.dumps(new_snapshot, indent=2, sort_keys=True) + "\n") + print(f"\nSnapshot written to {_SNAPSHOT}") + else: + print("\n(run with --write-snapshot to save this baseline)") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())