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()) 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/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_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 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