diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 76c31ed9..0347a764 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -29,8 +29,11 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv ## 2026-09-28 intake 与质量门 - `models`：生成与评审模型按 (provider, model, effort) 钉死（默认 `claude-code:claude-opus-5-5` - 与 `codex:gpt-6-sol:medium`，评审必须与生成不同家族）；不可用/超时/无法解析 → `ModelUnavailable`， - 调用方保持排队，绝不降级；每次调用经 recorder 记录输入、输出、用量。 + 与 `codex:gpt-6-sol:medium`，评审必须与主/备用生成器不同家族）；`KB_GENERATOR_FALLBACK` + 可显式配置一个备用生成器（如 `zcode:GLM-5.3`），主模型不可用/超时/无法解析时尝试一次。 + 评审、schema 修复及带花费阈值的调用均不切换（预算只为主模型预留，Zcode 无阈值能力）。 + 两个生成器都失败时保持排队。每次尝试经 recorder 记录实际模型、输入、输出、用量， + 备用调用带 `fallback_from`，intake/精炼/巡检变更集注明实际生成器，重建保留原归属。 - `sources`：知识仓库克隆的只读读取（`knowledge_files`、`external_texts`）、只读 GitHub 客户端 （合并 PR、PR 证据有界摘录、release/tag）、本机 Copilot 运行经验收件目录。 - `intake`：每个事件由生成器起草类型化操作（只允许 add/edit_same_meaning/replace/retire）， diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 5cbfa80b..4b3c4294 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -49,7 +49,18 @@ infermatrix-copilot kb calibrate --repo vllm-omni # 评审模 ``` 模型：`KB_GENERATOR`（默认 `claude-code:claude-opus-5-5`）、`KB_JUDGE`（默认 -`codex:gpt-6-sol:medium`）。每次模型调用写入 `$KB_STATE_DIR/traces/model_calls.jsonl`。 +`codex:gpt-6-sol:medium`）。生成器可显式配置一个备用模型： + +```bash +export KB_GENERATOR_FALLBACK=zcode:GLM-5.3 +``` + +主生成器缺少 CLI、未登录、调用失败、超时或回复无法解析时，同一次请求尝试备用生成器一次； +两者都不可用则事件继续排队。评审模型不切换，备用生成器也必须与评审来自不同家族。 +每次尝试记录实际 provider/model，备用调用还记录 `fallback_from`；变更集注明实际生成器。 +未设置此变量时只调用主模型。schema 错误仍由原有修复流程处理；带花费阈值的调用不切换， +因为预算只为主模型预留且 Zcode 无法执行阈值。 +调用记录位于 `$KB_STATE_DIR/traces/records/`。 运行经验（来源②）：`pr_debug` 的已验证修复记录经 `KB_BUGFIX_DIR`（同机，与运行侧 `KNOWLEDGE_INTAKE_DIR` 配为同一目录）与 `KB_BUGFIX_MAILBOX`（异机，`owner/repo#N`，只接受 `KB_BUGFIX_AUTHORS` 列出的作者）进入 intake。 diff --git a/src/infermatrix_copilot/kb_service/intake.py b/src/infermatrix_copilot/kb_service/intake.py index e67e8593..ebf2a626 100644 --- a/src/infermatrix_copilot/kb_service/intake.py +++ b/src/infermatrix_copilot/kb_service/intake.py @@ -65,6 +65,7 @@ class Draft: rationale: str = "" attempts: list[dict] = field(default_factory=list) rejected: bool = False # repairs exhausted, as opposed to "nothing to learn" + generator: str = "" # the successful model, including an explicit fallback @property def empty(self) -> bool: @@ -189,7 +190,8 @@ def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict, break else: if not operations: - return Draft([event_id], [], None, str(reply.data.get("rationale") or ""), attempts) + return Draft([event_id], [], None, str(reply.data.get("rationale") or ""), attempts, + generator=reply.role.label()) try: result = apply_operations(files, operations, release=release, today=today) except LifecycleError as exc: @@ -198,7 +200,8 @@ def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict, attempts.append({"attempt": attempt, "error": str(exc)}) continue accept_attempt(attempt) # only this call's reply became the change - return Draft([event_id], operations, result, str(reply.data.get("rationale") or ""), attempts) + return Draft([event_id], operations, result, str(reply.data.get("rationale") or ""), attempts, + generator=reply.role.label()) attempts.append({"attempt": attempt, "error": feedback.strip()}) return Draft([event_id], [], None, "rejected after repairs", attempts, rejected=True) diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index f74aecb0..89f37386 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -336,7 +336,7 @@ def to_people(reason: str) -> None: draft_keys=[], # a rebuild continues the change's refine lineage: the # limit never resets because context moved meanwhile - extra_detail={k: detail[k] for k in ("refine_round", "refine_history") + extra_detail={k: detail[k] for k in ("refine_round", "refine_history", "generator") if k in detail}) staged = rt.ledger.changeset(new_id) if staged["status"] not in ("gated", *IN_FLIGHT): diff --git a/src/infermatrix_copilot/kb_service/models.py b/src/infermatrix_copilot/kb_service/models.py index 5c8a8cb4..d27f2d5e 100644 --- a/src/infermatrix_copilot/kb_service/models.py +++ b/src/infermatrix_copilot/kb_service/models.py @@ -1,31 +1,39 @@ -"""Pinned model roles for the knowledge service, with no silent fallback. +"""Pinned model roles with an explicitly configured generator fallback. The generator drafts knowledge changes; the judge (a different model family) grades them. Both are pinned by (provider, model, reasoning effort) and -recorded on every call. An unavailable backend, a timeout, a call stopped -at its spend cap or an unparseable reply raises ``ModelUnavailable``; the caller leaves the work queued. It never -switches to a weaker model: a verdict must always say which model made it. +recorded on every call. ``KB_GENERATOR_FALLBACK`` may name one pinned model +to try when generation is unavailable. Judges never fall back, and schema +repair and spend limits retain their existing behavior. If both generators +are unavailable, the caller leaves the work queued. """ from __future__ import annotations import json +import logging import os import re import time -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, Callable GENERATOR_ENV = "KB_GENERATOR" # "provider:model[:effort]" +GENERATOR_FALLBACK_ENV = "KB_GENERATOR_FALLBACK" JUDGE_ENV = "KB_JUDGE" DEFAULT_GENERATOR = "claude-code:claude-opus-5-5" DEFAULT_JUDGE = "codex:gpt-6-sol:medium" _JSON_FENCE = re.compile(r"```(?:json)?\s*(\{.*\})\s*```", re.DOTALL) +logger = logging.getLogger(__name__) class ModelUnavailable(RuntimeError): """The pinned backend could not produce a usable reply; work stays queued.""" + def __init__(self, message: str, *, allow_fallback: bool = True): + super().__init__(message) + self.allow_fallback = allow_fallback + @dataclass(frozen=True) class ModelRole: @@ -33,6 +41,7 @@ class ModelRole: provider: str model: str effort: str = "" + fallback: ModelRole | None = None @classmethod def parse(cls, name: str, spec: str) -> "ModelRole": @@ -48,8 +57,16 @@ def label(self) -> str: def roles_from_env() -> tuple[ModelRole, ModelRole]: generator = ModelRole.parse("generator", os.environ.get(GENERATOR_ENV, DEFAULT_GENERATOR)) judge = ModelRole.parse("judge", os.environ.get(JUDGE_ENV, DEFAULT_JUDGE)) - if generator.provider == judge.provider and generator.model.split("-")[0] == judge.model.split("-")[0]: - raise ValueError("the judge must come from a different model family than the generator") + fallback_spec = os.environ.get(GENERATOR_FALLBACK_ENV, "").strip() + if fallback_spec: + fallback = ModelRole.parse("generator", fallback_spec) + if fallback.label().casefold() == generator.label().casefold(): + raise ValueError("the generator fallback must differ from the primary generator") + generator = replace(generator, fallback=fallback) + for candidate in (generator, generator.fallback): + if candidate is not None and candidate.provider == judge.provider \ + and candidate.model.casefold().split("-")[0] == judge.model.casefold().split("-")[0]: + raise ValueError("the judge must come from a different model family than every generator") return generator, judge @@ -108,6 +125,34 @@ def _transport(self, provider: str): def call_json(self, role: ModelRole, *, system: str, prompt: str, validate: Callable[[dict], None] | None = None, max_budget_usd: float | None = None, record_payload: bool = True) -> ModelReply: + """Try the primary, then one configured generator on unavailability. + + Both attempts use the same validation and payload policy. Calls with + a spend threshold and schema errors never trigger a second model call: + the caller has reserved for one model, and Zcode cannot enforce a cap. + """ + kwargs = dict(system=system, prompt=prompt, validate=validate, + max_budget_usd=max_budget_usd, record_payload=record_payload) + try: + return self._call_json(role, **kwargs) + except ModelUnavailable as primary_error: + if role.name != "generator" or role.fallback is None \ + or max_budget_usd is not None or not primary_error.allow_fallback: + raise + logger.warning("Knowledge generator %s unavailable; trying configured fallback %s", + role.label(), role.fallback.label()) + try: + return self._call_json(role.fallback, fallback_from=role.label(), **kwargs) + except ModelUnavailable as fallback_error: + raise ModelUnavailable( + f"{role.label()} unavailable: {primary_error}; " + f"fallback {role.fallback.label()} unavailable: {fallback_error}", + allow_fallback=False) from fallback_error + + def _call_json(self, role: ModelRole, *, system: str, prompt: str, + validate: Callable[[dict], None] | None = None, + max_budget_usd: float | None = None, record_payload: bool = True, + fallback_from: str = "") -> ModelReply: """``max_budget_usd`` is a per-call STOP THRESHOLD, not a hard cap: the transport starts no further API request once the call's spend reaches it, but the request that crosses it is billed in full. A @@ -115,20 +160,21 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str, request's worst case before calling. Requesting a threshold from a transport that cannot stop at one (``stops_at_spend`` False) is refused before dispatch; it is never approximated with ``max_tokens``.""" - transport = self._subscription_transports.pop(role.label(), None) - if transport is None: - transport = self._transport(role.provider) - cap: dict = {} - if max_budget_usd is not None: - if not max_budget_usd > 0: - raise ModelUnavailable(f"{role.label()}: max_budget_usd must be positive") - if not getattr(transport, "stops_at_spend", False): - raise ModelUnavailable(f"{role.provider} cannot stop a call at a spend threshold") - cap = {"max_budget_usd": max_budget_usd} started = time.time() identity = {"role": role.name, "requested": role.label(), "provider": role.provider, - "model": role.model, "effort": role.effort} + "model": role.model, "effort": role.effort, "fallback_from": fallback_from} try: + transport = self._subscription_transports.pop(role.label(), None) + if transport is None: + transport = self._transport(role.provider) + cap: dict = {} + if max_budget_usd is not None: + if not max_budget_usd > 0: + raise ModelUnavailable(f"{role.label()}: max_budget_usd must be positive", allow_fallback=False) + if not getattr(transport, "stops_at_spend", False): + raise ModelUnavailable(f"{role.provider} cannot stop a call at a spend threshold", + allow_fallback=False) + cap = {"max_budget_usd": max_budget_usd} reply = transport.complete( system=system, messages=[{"role": "user", "content": prompt}], model=role.model, effort=role.effort, role=role.name, **cap) @@ -140,6 +186,8 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str, "system": system if record_payload else "", "prompt": prompt if record_payload else "", "reply": "", "error": str(exc)[:2000] if record_payload else "transport failed (payload omitted)"}) + if isinstance(exc, ModelUnavailable): + raise raise ModelUnavailable(f"{role.label()} failed: {exc}") from exc seconds = time.time() - started text = "".join(getattr(block, "text", "") or "" for block in getattr(reply, "blocks", [])) @@ -160,7 +208,8 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str, failure: ModelUnavailable | None = None data: dict = {} if getattr(reply, "stop_reason", "") == "max_budget": - failure = ModelUnavailable(f"{role.label()} stopped at its spend threshold ${max_budget_usd}") + failure = ModelUnavailable(f"{role.label()} stopped at its spend threshold ${max_budget_usd}", + allow_fallback=False) elif getattr(reply, "stop_reason", "") == "max_tokens" or not text.strip(): failure = ModelUnavailable(f"{role.label()} timed out or returned nothing") else: @@ -170,7 +219,8 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str, try: validate(data) except Exception as exc: # any malformed shape is a controlled refusal - raise ModelUnavailable(f"{role.label()} reply failed its schema: {exc!r}") from exc + raise ModelUnavailable(f"{role.label()} reply failed its schema: {exc!r}", + allow_fallback=False) from exc except ModelUnavailable as exc: failure = exc if self._recorder is not None: diff --git a/src/infermatrix_copilot/kb_service/refine.py b/src/infermatrix_copilot/kb_service/refine.py index 42489fa3..2a6774bc 100644 --- a/src/infermatrix_copilot/kb_service/refine.py +++ b/src/infermatrix_copilot/kb_service/refine.py @@ -136,7 +136,8 @@ def to_people(why: str) -> None: external=external, operations=draft.operations, result=draft.result, evidence=[*evidence, mark], event_ids=[], release=release, draft_keys=[accepted_key(key, holder)] if accepted_key(key, holder) else [], - extra_detail={"refine_round": round_, "refine_history": history}) + extra_detail={"refine_round": round_, "refine_history": history, + "generator": draft.generator}) open_pr = changeset.get("pr_number") is not None rt.ledger.update_changeset(changeset["id"], status="superseding" if open_pr else "refined", detail={**detail, "refined_as": new_id, "refine_history": history, diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index 0c1b2697..70f5215a 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/src/infermatrix_copilot/kb_service/runtime.py @@ -132,7 +132,8 @@ def record(entry: dict) -> None: # (None when unset/unknown), on failures too result={"stop_reason": entry.get("stop_reason", ""), "max_budget_usd": entry.get("max_budget_usd"), - "cost_usd": entry.get("cost_usd")}, + "cost_usd": entry.get("cost_usd"), + **({"fallback_from": entry["fallback_from"]} if entry.get("fallback_from") else {})}, error=entry.get("error", "")) except Exception: # noqa: BLE001 - tracing never breaks a model call pass @@ -278,7 +279,8 @@ def _run_intake_locked(rt: KbRuntime, lifecycle: RepoLifecycle, *, max_events: i return gate_and_stage(rt, lifecycle, owner, kind="intake", base=base, base_sha=base_sha, external=external, operations=operations, result=result, evidence=evidence, event_ids=sorted(kept_ids), release=release, - draft_keys=[accepted[i] for i in sorted(kept_ids) if i in accepted]) + draft_keys=[accepted[i] for i in sorted(kept_ids) if i in accepted], + extra_detail={"generator": ", ".join(dict.fromkeys(d.generator for d in kept))}) def draft_key_for_event(repo: str, event_id) -> str: @@ -328,7 +330,8 @@ def gate_and_stage(rt: KbRuntime, lifecycle: RepoLifecycle, owner: str, *, kind: "operations": operations_json(operations), "event_ids": event_ids, **(extra_detail or {}), "base_sha": base_sha, "release": release, "decision": decision.to_dict(), - "generator": rt.generator.label(), "judge": rt.judge.label(), + "generator": (extra_detail or {}).get("generator") or rt.generator.label(), + "judge": rt.judge.label(), }) except Exception: rt.changeset_path(changeset_id).unlink(missing_ok=True) diff --git a/src/infermatrix_copilot/kb_service/sweep.py b/src/infermatrix_copilot/kb_service/sweep.py index 4f14dd1d..02601e93 100644 --- a/src/infermatrix_copilot/kb_service/sweep.py +++ b/src/infermatrix_copilot/kb_service/sweep.py @@ -187,7 +187,7 @@ def page_scope(files: dict[str, str], repo_dir: str, page: str) -> list[str]: def sweep_page(rt, lifecycle, *, page: str, files: dict[str, str], diff: str, hints: list[dict], - sweep: dict, release: str, today: str): + sweep: dict, release: str, today: str, generated_by: dict[str, str] | None = None): """Ask the generator about one page. Returns (operations, result), or None when every rule is kept; raises SweepPageFailed when no valid answer came.""" page_text = files[page] @@ -223,6 +223,8 @@ def sweep_page(rt, lifecycle, *, page: str, files: dict[str, str], diff: str, hi try: result = apply_operations(files, operations, release=release, today=today) accept_attempt(attempt) + if generated_by is not None: + generated_by[page] = reply.role.label() return operations, result except LifecycleError as exc: feedback = f"\n\nYour previous answer was rejected by the knowledge base: {exc}. Fix exactly that." @@ -310,6 +312,7 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, from ..trace_store import accepted_key, trace_context accepted: dict[str, str] = {} # page -> the accepted generator call's key + generated_by: dict[str, str] = {} """Run (or resume) one sweep. Progress is per page: a page is done once its answer is staged or it is kept; a page whose evaluation failed is retried on @@ -349,7 +352,7 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, with trace_context(draft_key=key, _accepted=holder, step="draft"): outcome = sweep_page(rt, lifecycle, page=page, files=base, diff=diff_cache[prefixes], hints=(audit_hints or {}).get(page, []), sweep=sweep, - release=release, today=today) + release=release, today=today, generated_by=generated_by) if accepted_key(key, holder): accepted[page] = accepted_key(key, holder) except SweepPageFailed as exc: @@ -380,7 +383,8 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, changeset_id = gate_and_stage(rt, lifecycle, owner, kind=kind, base=base, base_sha=base_sha, external=external, operations=operations, result=result, evidence=evidence, event_ids=[], release=release, hold=True, - draft_keys=[accepted[key]] if key in accepted else []) + draft_keys=[accepted[key]] if key in accepted else [], + extra_detail={"generator": generated_by.get(key, "")}) progress.setdefault("held", []).append(changeset_id) if key == "#t1": progress["t1_done"] = True diff --git a/test/test_kb_generator_fallback.py b/test/test_kb_generator_fallback.py new file mode 100644 index 00000000..bc2ecca2 --- /dev/null +++ b/test/test_kb_generator_fallback.py @@ -0,0 +1,183 @@ +"""Generator outages recover through an explicit, attributed subscription fallback.""" + +import json +from dataclasses import replace +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service.models import ( + ModelGateway, ModelRole, ModelUnavailable, roles_from_env, +) +from infermatrix_copilot.kb_service.replay import export_dataset +from infermatrix_copilot.kb_service.runtime import collect_events, run_intake, trace_recorder +from infermatrix_copilot.trace_store import TraceStore +from test_kb_intake_gate import GEN, JUDGE, _generator_then_judge, _runtime + +FALLBACK = ModelRole("generator", "zcode", "GLM-5.3") +PRIMARY = replace(GEN, fallback=FALLBACK) + + +class Transport: + stops_at_spend = True + + def __init__(self, answer): + self.answer = answer + self.calls = [] + + def complete(self, **kwargs): + self.calls.append(kwargs) + answer = self.answer(kwargs) if callable(self.answer) else self.answer + if isinstance(answer, Exception): + raise answer + return SimpleNamespace( + blocks=[SimpleNamespace(text=answer)], stop_reason="end_turn", + usage={}, model=kwargs["model"]) + + +def gateway(primary, fallback, *, recorder=None): + transports = {"claude-code": primary, "zcode": fallback} + return ModelGateway(None, transport_factory=transports.__getitem__, recorder=recorder) + + +@pytest.fixture(autouse=True) +def clean_model_env(monkeypatch): + for key in ("KB_GENERATOR", "KB_JUDGE", "KB_GENERATOR_FALLBACK"): + monkeypatch.delenv(key, raising=False) + + +def test_fallback_is_opt_in_and_keeps_codex_judge(monkeypatch): + generator, judge = roles_from_env() + assert generator == GEN and generator.fallback is None + monkeypatch.setenv("KB_GENERATOR_FALLBACK", "zcode:GLM-5.3") + generator, judge = roles_from_env() + assert generator == PRIMARY and judge == JUDGE + + +@pytest.mark.parametrize("spec,judge", [ + ("zcode", "codex:gpt-6-sol:medium"), + (GEN.label(), "codex:gpt-6-sol:medium"), + ("codex:gpt-6-mini:low", "codex:gpt-6-sol:medium"), +]) +def test_invalid_or_same_family_fallback_is_refused(monkeypatch, spec, judge): + monkeypatch.setenv("KB_GENERATOR_FALLBACK", spec) + monkeypatch.setenv("KB_JUDGE", judge) + with pytest.raises(ValueError): + roles_from_env() + + +@pytest.mark.parametrize("failure", [ + RuntimeError("account restricted"), + TimeoutError("timeout"), + "Your account is on hold and can't use Claude Code.", + "", +]) +def test_generator_outage_uses_exact_fallback_and_records_both_attempts(failure): + records = [] + primary, fallback = Transport(failure), Transport('{"operations": []}') + reply = gateway(primary, fallback, recorder=records.append).call_json( + PRIMARY, system="policy", prompt="evidence") + assert reply.role == FALLBACK and reply.served_model == "GLM-5.3" + assert len(primary.calls) == len(fallback.calls) == 1 + assert primary.calls[0]["messages"] == fallback.calls[0]["messages"] + assert [r["provider"] for r in records] == ["claude-code", "zcode"] + assert records[0]["error"] and not records[1]["error"] + assert records[1]["fallback_from"] == GEN.label() + + +def test_missing_cli_is_recorded_and_uses_fallback(): + records = [] + fallback = Transport('{"ok": true}') + + def factory(provider): + if provider == "claude-code": + raise ModelUnavailable("claude-code CLI is not installed") + return fallback + + g = ModelGateway(None, transport_factory=factory, recorder=records.append) + assert g.call_json(PRIMARY, system="s", prompt="p").role == FALLBACK + assert "not installed" in records[0]["error"] and len(fallback.calls) == 1 + + +def test_primary_success_never_calls_fallback(): + fallback = Transport(RuntimeError("must not be called")) + reply = gateway(Transport('{"ok": true}'), fallback).call_json(PRIMARY, system="s", prompt="p") + assert reply.role.label() == GEN.label() and not fallback.calls + + +def test_schema_failure_retains_existing_primary_repair_path(): + fallback = Transport('{"ok": true}') + + def validate(data): + if "ok" not in data: + raise ValueError("missing ok") + + with pytest.raises(ModelUnavailable, match="failed its schema"): + gateway(Transport('{}'), fallback).call_json(PRIMARY, system="s", prompt="p", validate=validate) + assert not fallback.calls + + +def test_fallback_reply_must_pass_the_same_validation(): + def validate(data): + raise ValueError("invalid operations") + + with pytest.raises(ModelUnavailable, match="invalid operations"): + gateway(Transport("account restricted"), Transport('{}')).call_json( + PRIMARY, system="s", prompt="p", validate=validate) + + +def test_thresholded_call_does_not_spend_a_second_reservation(): + fallback = Transport('{"ok": true}') + with pytest.raises(ModelUnavailable): + gateway(Transport("account restricted"), fallback).call_json( + PRIMARY, system="s", prompt="p", max_budget_usd=1) + assert not fallback.calls + + +def test_judge_never_falls_back_even_if_given_a_fallback_role(): + fallback = Transport('{"ok": true}') + role = replace(GEN, name="judge", fallback=FALLBACK) + with pytest.raises(ModelUnavailable): + gateway(Transport("account restricted"), fallback).call_json(role, system="s", prompt="p") + assert not fallback.calls + + +def test_payload_omission_applies_to_both_models(): + records = [] + gateway(Transport("account restricted"), Transport('{"ok": true}'), recorder=records.append).call_json( + PRIMARY, system="private policy", prompt="private evidence", record_payload=False) + assert len(records) == 2 + assert all(r["system"] == r["prompt"] == r["reply"] == "" for r in records) + + +def test_pending_event_recovers_and_change_and_trace_attribute_glm_with_codex_review(tmp_path): + primary = Transport("Your account is on hold and can't use Claude Code.") + unavailable = Transport(RuntimeError("zcode down")) + g = gateway(primary, unavailable) + rt, lifecycle = _runtime(tmp_path, g) + rt.generator = PRIMARY + store = TraceStore(tmp_path / "traces", environ={}) + rt.traces = store + g._recorder = trace_recorder(store) + collect_events(rt, lifecycle) + assert run_intake(rt, lifecycle) is None + assert len(rt.ledger.events("demo", "pending")) == 1 + answer = _generator_then_judge() + fallback = Transport(lambda kw: json.dumps(answer(FALLBACK, kw["messages"][0]["content"]))) + reviewer = Transport(lambda kw: json.dumps(answer(JUDGE, kw["messages"][0]["content"]))) + transports = {"claude-code": primary, "zcode": fallback, "codex": reviewer} + g._factory = transports.__getitem__ + changeset = rt.ledger.changeset(run_intake(rt, lifecycle)) + assert changeset["status"] == "gated" + assert changeset["detail"]["generator"] == FALLBACK.label() + assert changeset["detail"]["judge"] == JUDGE.label() + assert not rt.ledger.events("demo", "pending") and reviewer.calls + fallback_calls = [r for r in store.query(kind="model_call") if r["model"]["provider"] == "zcode"] + assert fallback_calls[-1]["model"]["served_model"] == "GLM-5.3" + assert fallback_calls[-1]["result"]["fallback_from"] == GEN.label() + out = tmp_path / "dataset.jsonl" + export_dataset(store, out, role="generator") + generators = [json.loads(line) for line in out.read_text().splitlines()] + assert len(generators) == 1 and generators[0]["model"]["provider"] == "zcode" + assert generators[0]["changeset_id"] == changeset["id"] + assert generators[0]["decision"]["status"] == "pass"