"""Bounded whole-suite orchestration with one shared deadline and cost allowance.""" from __future__ import annotations import hashlib import math import time import suite_backends from suite_assignments import decode from suite_contract import (EXECUTION_REVISION, MAX_BODY, MAX_RESULT, MODELS, PROMPT_REVISION, PROMPT_SHA256, REVISION, Problem, digest, encoded, reasoning_selection, validate_partition) from suite_policy import (BASE_NAME_LIMIT, MAX_GROUP, POLICY_REVISION, invocation, validate_natural) from suite_sizing import cap_families, review_summary MAX_PASSES = 5 def ordered_request(request, alternative=False): """Use stable independent orders while retaining every complete case record.""" def key(case): alias = case["alias"] return hashlib.sha256((POLICY_REVISION + ":order-b:" + alias).encode()).hexdigest() if alternative else alias cases = sorted(request["cases"], key=key) if alternative and len(cases) > 1 and [c["alias"] for c in cases] == sorted(c["alias"] for c in cases): cases.reverse() return {**request, "cases": cases} def capacity(call, provider, count, family_count=None): """Check the actual complete pass, including proposal/review input and schema.""" model = MODELS[provider] input_bytes = len(call["input"].encode()) + len(call["system"].encode()) + len(encoded(call["schema"])) if input_bytes > MAX_BODY: raise Problem("pass_request_too_large", 413, review_pass=call["stage"], input_bytes=input_bytes) if family_count is None: estimate = 1024 + 128 * count else: estimate = 1024 + 48 * count + 384 * family_count reserve = estimate + (8192 if provider == "claude" else 0) bound = input_bytes + model["overhead"] if reserve > model["output"] or bound + model["output"] * model.get("max_turns", 1) > model["context"]: raise Problem("pass_capacity", 422, review_pass=call["stage"], input_bytes=input_bytes, output_reservation_tokens=reserve) return {"input_bytes": input_bytes, "input_token_bound": bound, "input_token_count": None, "output_reservation_tokens": reserve, "output_reservation_verified": False, "input_count_method": "Complete UTF-8 input/system/schema byte bound plus harness overhead; not a tokenizer"} def preflight_workflow(request): """Choose once, respecting permissions; recheck actual expanded inputs at every pass.""" count = len(request["cases"]) candidates = ["local"] if request["routing"]["allow_external"]: candidates += request["routing"]["allowed_external_providers"] reasons = {} for provider in candidates: model = MODELS[provider] if not model["enabled"]: reasons[provider] = "unverified_output_capacity" continue try: initial = capacity(invocation("proposal_a", request), provider, count) # Even a minimum-size reconciliation must fit; its real proposals and # every subsequent request get checked again before any model call. capacity(invocation("reconciliation", request), provider, count, 1) except Problem as exc: reasons[provider] = exc.code continue return {"provider": provider, **model, **initial, **reasoning_selection(request, provider), "configuration_revision": REVISION, "prompt_revision": PROMPT_REVISION, "prompt_sha256": PROMPT_SHA256, "execution_revision": EXECUTION_REVISION, "policy_revision": POLICY_REVISION, "case_count": count, "source_sha256": digest(request["cases"]), "minimum_model_passes": 3, "maximum_model_passes": MAX_PASSES, "max_final_group_cases": MAX_GROUP, "max_final_name_characters": 64, "later_pass_capacity_verified": False, "later_pass_checks": "before_each_invocation"} raise Problem("capacity_or_unsupported_backend", 422, candidates=reasons) def sum_usage(records): """Aggregate measured counts only; keep an unknown counter unknown.""" values = [r.get("usage") for r in records] if not values or any(not isinstance(v, dict) for v in values): return None keys = set().union(*(v.keys() for v in values)) result = {} for key in keys: counts = [v.get(key) for v in values] if all(type(n) in (int, float) for n in counts): result[key] = sum(counts) elif all(isinstance(n, dict) for n in counts): result[key] = sum_usage([{"usage": n} for n in counts]) else: result[key] = None return result class Workflow: """Keep content in memory and allow only one pinned provider across model passes.""" def __init__(self, request, selected, cancel, client_ip, progress=None): self.request, self.provider, self.cancel = request, selected["provider"], cancel self.reasoning = selected.get("reasoning", MODELS[self.provider]["reasoning"]) self.client_ip, self.progress = client_ip, progress or (lambda _: None) self.started = time.monotonic() self.deadline = self.started + request["execution"]["max_seconds"] self.cost_limit, self.spent = request["execution"]["max_cost_usd"], 0.0 self.records = [] self.last_metadata = {} self.stage = "proposal_a" def checkpoint(self): """Fail closed on cancellation or exhaustion of the shared job budgets.""" if self.cancel.is_set(): raise Problem("cancelled", 409) if time.monotonic() >= self.deadline: raise Problem("job_time_budget_exhausted", 504) if self.spent > self.cost_limit: raise Problem("job_cost_budget_exhausted", 502) def call(self, stage, source, context=None, family_count=None): """Run one fresh complete-input invocation, retaining metadata before validation.""" self.stage = stage self.checkpoint() if len(self.records) >= MAX_PASSES: raise Problem("model_pass_limit", 502) call = invocation(stage, source, context) # Keep the preflight choice across independent ordering and larger review prompts. call["reasoning"] = self.reasoning limits = capacity(call, self.provider, len(source["cases"]), family_count) remaining_seconds = self.deadline - time.monotonic() remaining_cost = self.cost_limit - self.spent if remaining_cost <= 0: raise Problem("job_cost_budget_exhausted", 502) effective = {**source, "execution": {**source["execution"], "max_seconds": remaining_seconds, "max_cost_usd": remaining_cost}} started = time.monotonic() def report(activity=None): now = time.monotonic() self.progress({"current_pass": stage, "completed_model_passes": len(self.records), "maximum_model_passes": MAX_PASSES, "passes": self.records, "heartbeat_at": time.time(), "pass_elapsed_seconds": round(now - started, 1), "job_elapsed_seconds": round(now - self.started, 1), "job_remaining_seconds": round(max(0, self.deadline - now), 1), "cost_used_usd_estimate": round(self.spent, 8), "cost_limit_usd_estimate": self.cost_limit, **(activity or {})}) report() if self.provider == "claude": value, metadata = suite_backends.claude_generate(effective, self.cancel, invocation=call, progress=report) elif self.provider == "local": value, metadata = suite_backends.local_generate(effective, self.cancel, self.client_ip, invocation=call) else: raise Problem("unsupported_backend", 422) cost = metadata.get("cost_usd_estimate") if self.provider == "claude" else 0.0 record = {"stage": stage, "provider": self.provider, "model": MODELS[self.provider]["model"], "reasoning": self.reasoning, "wall_seconds": round(time.monotonic() - started, 3), **limits, "system_sha256": hashlib.sha256(call["system"].encode()).hexdigest(), "schema_sha256": digest(call["schema"]), "case_order_sha256": digest([c["alias"] for c in source["cases"]]), "allocated_seconds": remaining_seconds, "allocated_cost_usd": remaining_cost, "usage": metadata.get("usage"), "cli_turns": metadata.get("turns"), "duration_api_ms": metadata.get("duration_api_ms"), "cost_usd_estimate": cost, "cli_diagnostics": metadata.get("cli_diagnostics")} self.records.append(record) self.last_metadata = metadata if type(cost) not in (int, float) or not math.isfinite(cost) or cost < 0: raise Problem("budget_accounting_unavailable", 502) self.spent += cost self.checkpoint() report({"cli_running": False}) return decode(value, call, source) def execute(self): """Perform independent discovery, reconciliation, and bounded semantic reviews.""" source = ordered_request(self.request) a = self.call("proposal_a", source) validate_partition(a, source, name_limit=BASE_NAME_LIMIT, unique_names=False) b = self.call("proposal_b", ordered_request(source, True)) validate_partition(b, source, name_limit=BASE_NAME_LIMIT, unique_names=False) count = max(len(a["groups"]), len(b["groups"])) reconciled = self.call("reconciliation", source, {"proposal_a": a, "proposal_b": b}, count) validate_natural(reconciled, source) large = [g for g in reconciled["groups"] if len(g["members"]) > MAX_GROUP] reviewed = audited = None if large: reviewed = self.call("large_family_review", source, {"natural_partition": reconciled, "oversized_families": [g["members"] for g in large]}, len(reconciled["groups"])) validate_natural(reviewed, source, reconciled["groups"]) audited = self.call("decision_audit", source, {"original_partition": reconciled, "reviewed_partition": reviewed, "oversized_families": [g["members"] for g in large]}, len(reviewed["groups"])) validate_natural(audited, source, reconciled["groups"]) natural = audited or reconciled final, divisions = cap_families(natural, source) review = review_summary(a, b, reconciled, reviewed, audited, final, divisions) self.checkpoint() metadata = {**self.last_metadata, "passes": self.records, "reasoning": self.reasoning, "model_pass_count": len(self.records), "usage": sum_usage(self.records), "cli_diagnostics_scope": "last_model_pass", "duration_api_ms": sum(r["duration_api_ms"] for r in self.records) if all(type(r["duration_api_ms"]) in (int, float) for r in self.records) else None, "cost_usd_estimate": round(self.spent, 8), "turns": sum(r["cli_turns"] for r in self.records) if all(type(r["cli_turns"]) is int for r in self.records) else None, "natural_family_count": len(natural["groups"]), "final_task_count": len(final["groups"]), "singleton_statistics": {k: v for k, v in review["counts"].items() if "singleton" in k}, "policy_revision": POLICY_REVISION, "review_summary": review} if len(encoded({"result": final, **metadata})) > MAX_RESULT - 16384: raise Problem("response_too_large", 502) return final, metadata def generate(request, selected, cancel, client_ip, progress=None): """Never expose partial partitions as completion or broaden a failed route.""" workflow = Workflow(request, selected, cancel, client_ip, progress) try: return workflow.execute() except Problem as exc: details = {"failure_stage": "multi_pass_orchestration", **exc.details, "review_pass": workflow.stage, "completed_model_passes": len(workflow.records), "passes": workflow.records, "aggregate_usage": sum_usage(workflow.records + ([{"usage": exc.details["usage"]}] if isinstance(exc.details.get("usage"), dict) else [])), "aggregate_cost_usd_estimate": None if exc.code == "budget_accounting_unavailable" else round(workflow.spent, 8)} if exc.details.get("cost_usd_estimate") is not None: details["aggregate_cost_usd_estimate"] += exc.details["cost_usd_estimate"] raise Problem(exc.code, exc.status, **details) from None