"""Vertical-slice orchestrator + single-command entry (two-layer HITL wiring). ``run_project`` composes the whole method for ONE synthetic project on real (or injected) chat clients: 1. ``load_contracts`` — fail-fast: every config (incl. the verdict-feedback shape) is validated BEFORE any chat client is built. 2. load the project + retrieve cited chunks via the Step-7 data source -> first-class ``provenance.Citation`` list. 3. a FRESH maker-checker ``GroupChat`` debate (``fresh_workflow``), round-capped (``with_max_rounds``); **Layer-1 HITL** = the optional in-run ``with_request_info`` gate. 4. ``generate_via_llm`` -> blocking ``validate_proposal`` -> ``ValidatedProposal | Rejection`` (the token bound is the ``meter`` checked in the generate loop). 5. attach a first-class ``ProvenanceStamp``. 6. **Layer-2 (out-of-band)**: ``capture_verdict`` mints a stable id from the proposal's features; the decision/rationale come from ``verdict_input`` (function arg / CLI / fixture). The B11 expert notification is a STUB (``notify``) in Fase 2. 7. persist to the ``VerdictStore`` -> the next run's ``ExpeLContextProvider`` retrieval (the learning loop; this run also exercises the two-arg ``extend_instructions`` injection). The two HITL layers are deliberately distinct: **Layer-1** is the optional synchronous in-run review gate (no checkpoint — research 01: durable resume is fragile); **Layer-2** is the durable learned verdict captured out-of-band in the VerdictStore (D7-portable). """ from __future__ import annotations import asyncio import json from collections.abc import Callable, Iterable, Sequence from dataclasses import dataclass, replace from pathlib import Path from typing import Any, Literal, cast from agent_framework import BaseChatClient, SessionContext from pydantic import ValidationError from portfolio_optimiser.backends import Profile, get_backend, resolve_model from portfolio_optimiser.budget import ( Budget, BudgetMiddleware, BudgetRefused, PortfolioMeter, TokenMeter, ) from portfolio_optimiser.contracts import GoalConfig, GoalContract, load_contracts, load_goal_config from portfolio_optimiser.ledger import SavingsLedger, to_ore from portfolio_optimiser.datasource import ( bundle_citations, chunk_dict_to_citation, make_retrieval_tool, retrieve_chunks, ) from portfolio_optimiser.dimension import Dimension, admits, load_dimension from portfolio_optimiser.generate import generate_via_llm from portfolio_optimiser.ir import SavingsProposal from portfolio_optimiser.provenance import ProvenanceStamp from portfolio_optimiser.reference_domain import Project, load_reference_projects from portfolio_optimiser.validator import Rejection, ValidatedProposal, baseline_from_project from portfolio_optimiser import okf, outbox from portfolio_optimiser.semretrieval import ( SEMANTIC_WEIGHT_DEFAULT, Embedder, FakeEmbedder, HybridRanker, build_embedder, load_embedder_config, ) from portfolio_optimiser.verdicts import ( ExpeLContextProvider, ProposalFeatures, Verdict, VerdictStore, bundle_candidate_features, capture_verdict, load_verdicts_from_dir, similarity, ) from portfolio_optimiser.value_report import ( build_value_report, dump_report_json, format_report_text, ) from portfolio_optimiser.workflow import _MAKER_CHECKER_ROLES, fresh_workflow @dataclass(frozen=True) class RunResult: """The outcome of one project run: the validated/rejected proposal, its first-class provenance, the captured (Layer-2) verdict, the ExpeL hits surfaced for it, the store, the debate's converged output that the candidate was generated from (F1 traceability), and the checker's gate decision (Step 3/4: ``"approve" | "reject" | "absent"``). ``checker_verdict`` records the checker's decision distinctly from ``provenance.validator_decision`` so the two falsifiers (reasoning vs numbers) are never conflated.""" outcome: ValidatedProposal | Rejection provenance: ProvenanceStamp verdict: Verdict retrieved: list[Verdict] store: VerdictStore debate_output: str checker_verdict: str = "absent" @dataclass(frozen=True) class RunFailure: """One project that RAISED during a portfolio pass (S3.3, SC3 collect-and-continue). A DISTINCT type from ``RunResult`` rather than an error field on it, and deliberately so: the session spec's wording was "a ``RunResult`` slot with an error field", but ``RunResult`` is frozen with six required non-defaulted fields (``:89-94``) — a run that raised before producing an outcome has no honest value for ``outcome``, ``provenance`` or ``verdict``. Filling them with dummies would put FABRICATED provenance into the aggregate, which is the failure mode this repo's provenance rules exist to prevent. The exception is recorded as text (``error``) plus its class name (``error_type``) rather than the live exception object, so a ``PortfolioResult`` stays a plain frozen value with no traceback frames held alive.""" project_id: str error: str error_type: str @dataclass(frozen=True) class DryRunReport: """S4.2 offline ``--live-dry-run`` outcome (comparison protocol §4 pkt 3): everything a real run would use — profile, the resolved model-id per BUILT role, and the round/token parameters — captured WITHOUT a model call. A DISTINCT type from ``RunResult``, whose post-generation fields (outcome/provenance/verdict) do not exist yet on a run that stopped before the first model call.""" profile: str resolved_models: dict[str, str] max_rounds: int max_tokens: int top_k: int @dataclass(frozen=True) class GoalReached: """A savings-goal signal VALUE (Step 8, SC6) — NOT an exception. Structured like ``BudgetExceeded`` (``budget.py:22-34``) but semantically SUCCESS (the goal was reached), not resource exhaustion (H1). Used as a ``stop_reason`` value + a loop ``break``, never ``raise``d. ``scope`` is ``"portfolio"`` (the whole pass) or ``"project"`` (one pid); ``limit_ore`` is the threshold that was met, ``observed_ore`` the accumulated realized sum that met it (``>=``).""" scope: Literal["project", "portfolio"] project_id: str | None limit_ore: int observed_ore: int @dataclass(frozen=True) class BudgetStop: """A GLOBAL token-cap stop signal VALUE (S3.4/F10) — NOT an exception, and NOT a goal. Structured like ``GoalReached`` and carried the same way (a ``stop_reason``-shaped value plus a loop ``break``), but it is kept as its OWN field rather than widening ``stop_reason``: the two stops mean opposite things. A goal-stop is success (the savings target was met); this is resource exhaustion (the pass ran out of tokens). Folding them into one field would let a caller read "we stopped" without being able to tell which happened. ``required_tokens`` is what one more run would have needed; ``remaining_tokens`` is what was actually left. Both are recorded because their DIFFERENCE is the operator's next decision.""" limit_tokens: int spent_tokens: int remaining_tokens: int required_tokens: int @dataclass(frozen=True) class PortfolioResult: """The outcome of a sequential fan-out over N projects (SC2). ``runs`` is one ``RunResult`` per project in input order; ``store`` is the ONE shared ``VerdictStore`` threaded across every run (the cross-project ExpeL learning loop). The remaining fields are a thin aggregate over ``runs``: ``validated_count`` / ``rejected_count`` partition the outcomes; ``sum_claimed_saving_nok`` totals the claimed saving of the validated proposals only; ``sum_token_usage`` totals every run's provenance token usage. ``stopped_early`` / ``stop_reason`` record a Step-8 goal-stop, ``failures`` records the projects that RAISED (S3.3 collect-and-continue), and ``budget_stop`` records a S3.4 global-token-cap stop (which also sets ``stopped_early``, but is kept apart from ``stop_reason`` because exhaustion is not success): all four default so the frozen aggregate and every existing constructor call are unaffected. ``runs`` and ``failures`` PARTITION the projects that were actually submitted — a project appears in exactly one of them, never both, and the counts do not overlap. ``validated_count`` / ``rejected_count`` therefore total ``len(runs)``, not the portfolio size: a failure is neither a validation nor a rejection, and folding it into either would misreport the pass.""" runs: tuple[RunResult, ...] store: VerdictStore validated_count: int rejected_count: int sum_claimed_saving_nok: float sum_token_usage: int stopped_early: bool = False stop_reason: GoalReached | None = None failures: tuple[RunFailure, ...] = () budget_stop: BudgetStop | None = None def _authored_texts(result: Any, name: str) -> list[str]: """The texts of ``get_outputs()`` entries authored by participant ``name`` (proposer/checker), in surfaced order. MAF surfaces ``author_name`` on each output's ``messages`` — NOT on the ``AgentResponse`` itself (verified against 1.9.0) — so we match through ``messages``. This separates the proposer's converged output (fed to generation, F1) from the checker's gate verdict (Step 3/4); the orchestrator's termination notice is authored by neither, so it is excluded automatically.""" texts: list[str] = [] for out in result.get_outputs(): if not any(getattr(m, "author_name", None) == name for m in getattr(out, "messages", [])): continue text = out if isinstance(out, str) else getattr(out, "text", None) if text: texts.append(text) return texts def _debate_text(result: Any) -> str: """The PROPOSER's converged output (fed into generation, F1). With ``output_from=agents`` both participants surface, so we select proposer-authored outputs specifically — taking the last of ALL surfaced outputs would feed the checker's verdict to generation at even round counts. Returns ``""`` when the proposer produced no surfaced text.""" proposer_texts = _authored_texts(result, "proposer") return proposer_texts[-1] if proposer_texts else "" def _checker_verdict(result: Any) -> tuple[str, str]: """Parse the checker's gate verdict from its surfaced debate output (Step 3/4, målbilde §2/§6). Returns ``(decision, reason)``: ``"reject"`` ONLY on an explicit ``VERDICT: REJECT`` (with the trailing reason), ``"approve"`` on an explicit ``VERDICT: APPROVE``, else ``"absent"``. The gate is opt-in-reject (fail-open): a missing/unparseable marker never blocks, so the deterministic validator stays the sole gate on those runs.""" checker_texts = _authored_texts(result, "checker") text = checker_texts[-1] if checker_texts else "" upper = text.upper() if "VERDICT: REJECT" in upper: reason = text[upper.index("VERDICT: REJECT") + len("VERDICT: REJECT") :] return "reject", reason.lstrip(" -:—").strip() if "VERDICT: APPROVE" in upper: return "approve", "" return "absent", "" def _project_by_id(project_id: str) -> Project: for project in load_reference_projects(): if project.id == project_id: return project raise ValueError(f"unknown project_id: {project_id!r}") def _project_from_bundle( bundle_dir: str, project_id: str, *, bundle: okf.Bundle | None = None ) -> Project: """Derive a minimal ``Project`` from an OKF bundle (so a bundle the loop runs need NOT be a road reference-domain project). Only ``id`` + ``name`` reach the generation prompt (``generate._build_messages``), so ``cost_items`` is empty and ``verdict_input`` is unused here (the Layer-2 decision flows via the ``verdict_input`` argument). Fail-fast: the bundle's IR ``project_id`` must match the requested id. ``bundle`` reuses an already-navigated bundle to avoid a second navigation.""" ir = okf.load_ir_projection(bundle_dir) if ir["project_id"] != project_id: raise ValueError(f"bundle project_id {ir['project_id']!r} != requested {project_id!r}") nav = bundle if bundle is not None else okf.navigate_bundle(bundle_dir) project_file = next((f for f in nav.files if f.type == "project"), None) name = ( project_file.frontmatter.get("title", project_id).strip('"') if project_file is not None else project_id ) return Project( id=project_id, name=name, description="", currency="NOK", cost_items=(), docs_dir=bundle_dir, verdict_input={}, ) def _features_of(proposal: SavingsProposal) -> ProposalFeatures: return ProposalFeatures( affected_codes=frozenset(item.code for item in proposal.affected_items), measure_type=proposal.measure, claimed_saving_nok=proposal.claimed_saving_nok, description=proposal.measure, ) def _default_factory(profile: Profile | str) -> Callable[[str], BaseChatClient]: def factory(role: str) -> BaseChatClient: return get_backend(profile).create_chat_client(model=resolve_model(profile, role)) return factory async def run_project( project_id: str, profile: Profile | str = Profile.LOCAL, *, docs_dir: str, verdict_input: dict[str, str], bundle_dir: str | None = None, dimension: Dimension | None = None, store: VerdictStore | None = None, verdict_dir: str | None = None, outbox_dir: str | None = None, run_id: str | None = None, client_factory: Callable[[str], BaseChatClient] | None = None, max_rounds: int = 3, max_tokens: int = 100_000, top_k: int = 3, enable_layer1_hitl: bool = False, notify: Callable[[Verdict], None] | None = None, meter: TokenMeter | None = None, live_dry_run: bool = False, semantic_retrieval: bool = False, embedder: Embedder | None = None, ) -> RunResult | DryRunReport: """Run the vertical slice for ONE project. ``client_factory`` is the test-injection seam (defaults to the real backend). ``verdict_input`` carries the expert decision/rationale (Layer-2). ``bundle_dir`` (Fase 2a) makes the run OKF-bundle-driven: the project is derived from the bundle and, before generation, the candidate's prior verdicts in ``store`` are folded into the hypothesis prompt (Step-1 ExpeL wiring, målbilde §5/§7). ``verdict_dir`` (Fase 5, Steg 7, målbilde §3/§7) is the async file inbox: a folder of expert/persona-authored verdict files (plain JSON, R2 raw layer) MERGED into the store BEFORE the Step-1 fold, so a verdict dropped after an earlier run is consumed by this separate, later run — the long feedback loop, fully resumable across runs separated in time. The system READS this folder; it does not write to it (the expert/persona writes, målbilde §3). ``outbox_dir`` (Fase 2a, Steg 7 output layer, målbilde §3, R2) is the RAW OUTBOX: when set, the run's proposal + outcome artefacts are written there via ``outbox.write_outbox`` (``run_id`` is then REQUIRED — no wall-clock/uuid default, for byte-determinism). The outbox dir should be DISTINCT from any ``verdict_dir`` inbox: writing the outbox into a folder later read as an inbox would re-ingest raw agent output and bypass the Step-8 promotion gate (self-contamination) — documented here, not enforced. Raises ``pydantic.ValidationError`` on a bad contract and ``BudgetExceeded`` when the token/round cap is crossed, and ``ValueError`` when ``outbox_dir`` is set without a ``run_id``. ``live_dry_run`` (S4.2, comparison protocol §4 pkt 2/3) is the offline drill: it walks the whole path up to the EAGER client build, writes the run-config artefact (when ``outbox_dir`` is set), and returns a ``DryRunReport`` BEFORE the first model call (``debate.run``) — zero chat calls. ``semantic_retrieval`` (S3.1) is the opt-in scaling SEAM — the deliverable is the extension point, not better retrieval. When true, a ``HybridRanker`` blends a cosine term over the embedded feature triple (sorted cost codes, measure type, magnitude bucket) with the structural score, which lets a prior verdict on a DIFFERENT cost-code set outrank one that ties structurally. The shipped ``FakeEmbedder`` is a deterministic sha256 projection carrying NO semantics, so over a structural tie the resulting order is deterministic but arbitrary; retrieval *quality* arrives only with an embedder injected via ``embedder=`` or ``--embedder-config``. Default false keeps the structural ranking exactly as before.""" # 0. Fail-fast: an outbox write is byte-deterministic and keyed on run_id — no wall-clock default. if outbox_dir is not None and run_id is None: raise ValueError( "run_id is required when outbox_dir is set (no wall-clock/uuid default — the outbox " "artefacts are byte-deterministic and keyed on run_id)" ) # 1. Fail-fast: validate ALL contracts (incl. the verdict-feedback shape) before any client. load_contracts( {"docs_dir": docs_dir, "top_k": top_k}, {"max_rounds": max_rounds, "max_tokens": max_tokens}, verdict_input, ) # 1b. Long loop (Steg 7): ingest the async verdict inbox INTO the store before the Step-1 fold. # Merge (not replace) into the passed store so run_portfolio's cross-project threading stays # intact; store.add is idempotent on the content-hash id. A verdict that landed after an earlier # run thus reaches THIS run's hypothesis via the existing fold below — no change to the fold. if verdict_dir is not None: store = store if store is not None else VerdictStore(verdicts=[]) for dropped in load_verdicts_from_dir(verdict_dir): store.add(dropped) # 2-3. Project + agent read-context + first-class citations. A bundle run derives ALL THREE from # the navigated OKF bundle via progressive disclosure (verdict layer EXCLUDED — målbilde §2/§4), # NOT keyword chunk-stuffing; the road path keeps the chunk-retrieval data source. ``debate_tools`` # is the query-time retrieval surface — empty on the bundle path (navigation already placed the # curated context in the prompt, and a docs_dir==bundle_dir tool would re-leak the verdict layer). # S4.0 (F3): the run path SETS the validator's cost baseline, so the deterministic gate is # anchored to the project's real cost lines instead of the ones the proposal asserts. # * road path: the reference project's own ``cost_items`` ARE the baseline -> always anchored. # * bundle path: anchored only when the bundle SHIPS a ``cost-baseline.json``. A bundle written # before the amendment (every commons-owned golden) is legitimately un-anchored -> None = # pre-S4.0 behaviour. A baseline that exists but is malformed still raises (fail-closed). if bundle_dir is not None: bundle = okf.navigate_bundle(bundle_dir) project = _project_from_bundle(bundle_dir, project_id, bundle=bundle) baseline = okf.load_optional_cost_baseline(bundle_dir) # §4.1a context-scope: agents read ONLY dimension-scoped bundle knowledge (Step-3 filter); # dimension=None keeps the full context, byte-identical to before. context = okf.bundle_context(bundle, dimension=dimension.id if dimension else None) citations = bundle_citations(bundle) debate_tools: list[Any] = [] else: project = _project_by_id(project_id) baseline = baseline_from_project(project) chunks = retrieve_chunks("cost saving measure", docs_dir, top_k) citations = [chunk_dict_to_citation(c) for c in chunks] context = "\n".join(c["snippet"] for c in chunks) debate_tools = [make_retrieval_tool(docs_dir, top_k=top_k)] if not citations: raise ValueError(f"no citable content in docs_dir: {docs_dir!r}") # 4. Budget + maker-checker debate (round-capped; Layer-1 HITL optional). The shared meter is # driven on the debate's chat calls by the BudgetMiddleware (the brief's named short-circuit # mechanism); ``debate_tools`` exposes the citation-bearing data source on the road path. meter = ( meter if meter is not None else TokenMeter(Budget(max_tokens=max_tokens, max_rounds=max(max_rounds * 4, 4))) ) factory = client_factory if client_factory is not None else _default_factory(profile) budget_mw = BudgetMiddleware(meter) debate = fresh_workflow( factory, max_rounds=max_rounds, enable_layer1_hitl=enable_layer1_hitl, tools=debate_tools, middleware=[budget_mw], ) # S4.2 cut (comparison protocol §4 pkt 2/3): everything above is offline — contracts, budget, and # the EAGER client build (fresh_workflow constructs the proposer+checker clients, workflow.py:64). # Capture the run-config (resolved model per BUILT role, profile, params, token cap) and, for a # ``--live-dry-run``, STOP HERE — before the first (paid) model call at ``debate.run`` below. if outbox_dir is not None or live_dry_run: # ``resolved_models`` reflects the configured MAP (the default factory's model-ids for the M2 # run). Under an injected ``client_factory`` the built clients may differ (e.g. "synthetic"); # ``provenance.model`` (below) stays the authority on the client actually built. resolved_models = {role: resolve_model(profile, role) for role in _MAKER_CHECKER_ROLES} if outbox_dir is not None: assert run_id is not None # narrowed by the step-0 guard (no wall-clock default) outbox.write_run_config( outbox_dir, run_id, profile=Profile(profile).value, resolved_models=resolved_models, max_rounds=max_rounds, max_tokens=max_tokens, top_k=top_k, ) if live_dry_run: return DryRunReport( profile=Profile(profile).value, resolved_models=resolved_models, max_rounds=max_rounds, max_tokens=max_tokens, top_k=top_k, ) result = await debate.run(f"Find a cost-saving measure for {project.id}.\nContext:\n{context}") # F1: the candidate must derive from the DEBATE. Feed the proposer's converged output into # generation (retrieval context is the last-resort fallback only). The checker's verdict # (Step 3/4) is parsed from the SAME debate result and gates the outcome below. debate_output = _debate_text(result) checker_decision, checker_reason = _checker_verdict(result) gen_context = debate_output or context # Step-1 ExpeL wiring (Fase 2a, målbilde §5/§7): fold the candidate's prior verdicts INTO the # hypothesis context BEFORE generation, keyed on the OKF bundle's candidate features (available # pre-hypothesis). THIS is the one missing dataflow — previously ExpeL was computed # post-generation into a discarded SessionContext (step 7 below), so a prior verdict could not # reach the next hypothesis. Bundle-driven path with a populated store only. # # Scope of the --semantic-retrieval opt-in, stated precisely (an earlier version of this # comment claimed "the road path is untouched", which the flag made false): the ranker built # below is passed to ALL THREE retrievals this run performs — this fold, and the post-hoc # ExpeLContextProvider + store.retrieve in step 7 — so the flag reaches the road path's # proposal-keyed retrieval too. What IS untouched on the road path is the fold itself: it stays # bundle-gated, so a --docs-dir-only run remains single-shot either way. # S3.1 opt-in: build the hybrid ranker as a LOCAL, then pass it explicitly at each retrieval # this run performs. It is deliberately not assigned to ``store.retriever``: the store is # caller-owned (``run_portfolio`` threads one store across every project, and a library caller # may reuse theirs), so a store-global assignment leaked this run's opt-in into every later use # of that object — including a subsequent run with the flag OFF. Flag off => ranker stays None # => ``retrieve`` falls through to the StructuralRetriever default. ranker = ( HybridRanker( embedder if embedder is not None else FakeEmbedder(), similarity, SEMANTIC_WEIGHT_DEFAULT, ) if semantic_retrieval else None ) if bundle_dir is not None and store is not None and store.verdicts: expel_query = bundle_candidate_features(bundle_dir) fewshot = ExpeLContextProvider( store, expel_query, k=top_k, retriever=ranker ).format_fewshot() gen_context = f"{fewshot}\n\n{gen_context}" # 5. Structured candidate -> blocking validation on the NUMBERS; token bound = the meter. proposer_client = factory("proposer") validator_outcome = await generate_via_llm( proposer_client, project, gen_context, meter, baseline=baseline ) proposal = validator_outcome.proposal # 6. First-class provenance stamp (authoritative; independent of MAF Annotation). # F1: an injected client_factory stamps the injected client's REAL model ("unknown" is the # neutral fallback for a client that doesn't surface one -- never a fabricated name); the # default path keeps the deterministic resolve_model. validator_decision reflects the VALIDATOR # (the numbers) ONLY -- stamped from validator_outcome BEFORE the checker override below, so a # checker-gated proposal whose numbers passed is never mislabelled as validator-rejected. model = ( (getattr(proposer_client, "model", None) or "unknown") if client_factory is not None else resolve_model(profile, "proposer") ) stamp = ProvenanceStamp( citations=citations, model=model, role="proposer", validator_decision=( "validated" if isinstance(validator_outcome, ValidatedProposal) else "rejected" ), token_usage=meter.tokens, ) # 6b. Step 3/4 checker gate (målbilde §2/§6): the validator falsifies the numbers, the checker # falsifies the reasoning. An explicit checker REJECT blocks an otherwise-validated proposal; a # validator rejection (the numbers) already stands. Fail-open: APPROVE/absent never blocks. outcome: ValidatedProposal | Rejection if isinstance(validator_outcome, ValidatedProposal) and checker_decision == "reject": outcome = Rejection(proposal=proposal, reason=f"checker rejected: {checker_reason}") else: outcome = validator_outcome # 6c. Step 2 dimension scope gate (§4.1b): a candidate whose measure_type/codes fall OUTSIDE the # run's dimension is rejected. A scope/type gate placed AFTER the checker override (preserves # test_checker_gate_loadbearing) — NOT a new numeric gate: validate_proposal stays the only # blocking numeric gate and provenance.validator_decision (the numbers) is untouched. Mirrors the # override form: only an otherwise-standing ValidatedProposal can be flipped to a Rejection. if dimension is not None and isinstance(outcome, ValidatedProposal): feats = _features_of(proposal) if not admits( measure_type=feats.measure_type, codes=feats.affected_codes, dimension=dimension ): outcome = Rejection( proposal=proposal, reason=f"outside dimension {dimension.id!r}: measure_type={feats.measure_type!r}", ) # 7. ExpeL (regression guard + traceability): exercises the two-arg extend_instructions # injection on a REAL SessionContext (the Critical Fase-1 GA-signature guard), and surfaces # the proposal-keyed retrieval for RunResult.retrieved. On the bundle path the load-bearing # ExpeL->prompt dataflow already happened pre-generation (above); this block's SessionContext # is NOT what reaches the prompt. store = store if store is not None else VerdictStore(verdicts=[]) features = _features_of(proposal) provider = ExpeLContextProvider(store, features, k=top_k, retriever=ranker) sctx = SessionContext(input_messages=[], instructions=[]) await provider.before_run(agent=None, session=None, context=sctx, state={}) retrieved = store.retrieve(features, k=top_k, retriever=ranker) if store.verdicts else [] # 8. Layer-2 (out-of-band): capture the durable verdict + persist; B11 notify is a stub. verdict = capture_verdict(features, verdict_input["decision"], verdict_input["rationale"]) store.add(verdict) if notify is not None: notify(verdict) # S2.1 outbox (RAW output layer, målbilde §3): persist the run's proposal + outcome artefacts # when configured. Wired ONLY here — no new consumer (S5.1/S5.2 are Non-Goals this bolk). run_id # is guaranteed non-None by the fail-fast guard at the top. if outbox_dir is not None: assert run_id is not None # narrowed by the step-0 guard; keeps the type checker honest outbox.write_outbox( outbox_dir, run_id, outcome=outcome, provenance=stamp, checker_verdict=checker_decision, verdict_id=verdict.id, ) return RunResult( outcome=outcome, provenance=stamp, verdict=verdict, retrieved=retrieved, store=store, debate_output=debate_output, checker_verdict=checker_decision, ) def _aggregate(runs: tuple[RunResult, ...], store: VerdictStore) -> PortfolioResult: """Thin aggregate over the per-project runs (SC2): partition validated/rejected, total the validated claimed saving, and total every run's provenance token usage.""" validated = [r for r in runs if isinstance(r.outcome, ValidatedProposal)] rejected = [r for r in runs if isinstance(r.outcome, Rejection)] return PortfolioResult( runs=runs, store=store, validated_count=len(validated), rejected_count=len(rejected), sum_claimed_saving_nok=sum(r.outcome.proposal.claimed_saving_nok for r in validated), sum_token_usage=sum(r.provenance.token_usage for r in runs), ) def _baseline_ore(projects: Iterable[Project]) -> int: """Addressable baseline in øre, quantized PER COST LINE and summed as integers (Kø-(p)). Both sides of the goal comparison must be computed in the same order. ``observed_ore`` is ``SavingsLedger``'s sum of per-candidate integer øre; a baseline that summed ``Project.total_cost`` floats and quantized the total ONCE put the threshold on a different scale — three ``60000.005`` NOK lines are ``18000003`` øre per line but ``18000001`` summed first, enough to flip a percent goal. Each cost line is a real amount, so the per-line value is the one that exists; integer addition also keeps the total order-independent, which ``Project.total_cost``'s float fold is not under the D-D wave model.""" return sum(to_ore(item.total_cost) for p in projects for item in p.cost_items) def _goal_limit_if_reached(goal: GoalContract, observed_ore: int, baseline_ore: int) -> int | None: """The threshold ``observed_ore`` MET (``>=``), or ``None`` if the goal is not yet reached. An absolute-øre target compares directly; a percent target is taken against ``baseline_ore`` (the addressable cost in øre). When both are set, reaching EITHER counts as met.""" if goal.absolute_ore is not None and observed_ore >= goal.absolute_ore: return goal.absolute_ore if goal.percent is not None: if baseline_ore <= 0: raise ValueError( f"percent goal ({goal.percent}%) is meaningless against a non-positive baseline " f"({baseline_ore} øre): int(percent/100 * 0) == 0 would falsely read as 'goal " "reached'. Supply an absolute_ore target, or ensure the addressable baseline is > 0." ) threshold = int(goal.percent / 100 * baseline_ore) if observed_ore >= threshold: return threshold return None def _wave_snapshot(store: VerdictStore) -> VerdictStore: """A per-project copy of the shared store's CURRENT verdicts (D-D wave model). Two coroutines appending to one list would interleave by completion order, which no barrier could then undo. Giving every project in a wave its own copy removes the race at its source rather than serializing it away with a lock — a lock would order the appends by whoever won, which is exactly the nondeterminism being eliminated. ``retriever`` is carried across deliberately: it is the S3.1 opt-in seam, and a snapshot that dropped it would silently downgrade a caller-owned store's semantic retrieval to the structural default mid-pass. **Field-complete by construction.** ``dataclasses.replace`` carries over every field ``VerdictStore`` declares and overrides only ``verdicts``, so a field added later is copied without this function being touched. The hand-enumerated version this replaced could silently drop one — the exact defect it was first written with (the ``retriever`` omission), which its own docstring then recorded as a standing hazard. Deriving the copy from the dataclass removes the hazard instead of documenting it, and does so without the field-count assertion that idea was rejected for: there is nothing left to keep in sync. That docstring credited a Step-2 contract test with catching the omission. MEASURED while this change was made: no such coverage remained — reinstating the hand-enumerated form left the whole suite green, so the S3.1 retriever seam could be downgraded mid-pass in silence. The gate is now ``test_wave_snapshot_carries_every_field_except_the_copied_verdicts``. ``verdicts`` is still listed explicitly, and must be: ``replace`` copies field REFERENCES, so omitting it would hand back a store sharing the caller's list — the very race this snapshot exists to remove, reintroduced by the call that looks tidiest.""" return replace(store, verdicts=list(store.verdicts)) def _merge_wave(store: VerdictStore, wave: Sequence[tuple[str, VerdictStore]]) -> None: """The deterministic merge barrier — the seam S3.3 rests on. Each project ran against its own snapshot, so its NEW verdicts are the ones absent from the wave-start store. They are merged back in **wave-submission order** — the order the caller listed the projects in — so the shared store's contents depend on the portfolio's membership and never on which project's model round-trips happened to finish first. **Submission order, NOT lexicographic project_id.** The plan specified a sort on ``project_id``; the Step-2 contract test measured that it produces a deterministic order which is nevertheless the WRONG one. On the shipped fixture, lexicographic order is BRU/FV42/RV13 while the sequential pass yields FV42/RV13/BRU, so a ``project_id`` sort satisfies "deterministic" while breaking "identical to ``concurrency=1``" — and the second is the actual contract. ``wave`` arrives in submission order, so preserving it is the fix. **Detach point: this function's ordering discipline is only half the seam — see ``_wave_snapshot``, which is the half that carries the load.** Because each project writes to its own copy, ``wave`` is never reordered by completion, so iterating it in order is already deterministic. Removing the SNAPSHOT is what turns the shared list back into a race and takes ``test_concurrent_pass_is_byte_identical_to_sequential`` RED. This is recorded plainly rather than dressing the loop in a ``sorted(...)`` that would re-sort an already-ordered list and read as a guard while guarding nothing. Safe to apply as-is because ``VerdictStore.add`` is first-write-wins per content-hash id (``verdicts.py:303-304``) — the wave-start verdicts every snapshot carries are re-offered and dropped, so only the new ones land.""" seen = {v.id for v in store.verdicts} for _pid, snapshot in wave: for verdict in snapshot.verdicts: if verdict.id not in seen: store.add(verdict) seen.add(verdict.id) def _waves(ids: list[str], k: int) -> list[list[str]]: """Partition ``ids`` into consecutive waves of at most ``k``, preserving caller order (D-D). Order preservation is the whole point: the wave model may change WHEN a project runs, never WHICH projects run nor in what sequence they are submitted. At ``k=1`` every wave holds a single project, so the wave loop degenerates to the pre-S3.3 sequential order by construction — that is what makes the parameter behaviour-preserving at its default.""" return [ids[i : i + k] for i in range(0, len(ids), k)] def _run_meter( meter_factory: Callable[[], TokenMeter] | None, portfolio_meter: PortfolioMeter | None, max_rounds: int, ) -> TokenMeter | None: """The per-project meter ``run_portfolio`` hands to one run: the injected test seam, a meter BOUND to the portfolio ledger, or ``None`` (letting ``run_project`` build its own, unchanged). The bound meter mirrors ``run_project``'s own construction (``max(max_rounds * 4, 4)``) so the only difference the portfolio cap introduces is the token ceiling and the shared ledger — the round budget is not quietly redefined along the way.""" if meter_factory is not None: return meter_factory() if portfolio_meter is None: return None return TokenMeter( Budget( max_tokens=portfolio_meter.budget.max_tokens_per_run, max_rounds=max(max_rounds * 4, 4), ), portfolio=portfolio_meter, ) async def run_portfolio( project_ids: Sequence[str] | None = None, profile: Profile | str = Profile.LOCAL, *, dimension: Dimension | None = None, ledger: SavingsLedger | None = None, goals: GoalConfig | None = None, store: VerdictStore | None = None, client_factory: Callable[[str], BaseChatClient] | None = None, max_rounds: int = 3, max_tokens: int = 100_000, top_k: int = 3, concurrency: int = 1, meter_factory: Callable[[], TokenMeter] | None = None, portfolio_meter: PortfolioMeter | None = None, semantic_retrieval: bool = False, embedder: Embedder | None = None, ) -> PortfolioResult: """Fan out over a portfolio of independent projects SEQUENTIALLY, composing ``run_project`` as-is (every project's execution state — meter, debate, retrieval context — is built fresh per call, so sequential reuse is inherently isolated). ONE ``VerdictStore`` is threaded across every run, AND each project's ``bundle_dir``/``verdict_dir`` are threaded into ``run_project`` (Fase 2a S2.0): a bundle-backed project's ``bundle_dir``-gated Step-1 ExpeL fold then folds the store's prior verdicts into its hypothesis prompt, so a verdict captured on project k materially reaches project k+1's hypothesis — the cross-project learning loop is delivered, not merely asserted. ``verdict_dir`` also lets each project merge its async file inbox (Steg 7) on the portfolio path. ``project_ids`` defaults to every loaded project; an unknown id raises ``ValueError``. ``meter_factory`` (test seam) supplies a per-project meter — inject a shared meter to make the isolation guard go red (SC3). Step 8 goal-stop (SC6): ``ledger`` carries EARLIER, out-of-band HITL realizations (the long file loop / Steg-7 role split — ``run_project`` never realizes mid-pass, C3), so the check reads an ACCUMULATED sum, it does not build one during the pass. BEFORE running each pid, the accumulated realized sum is compared to ``goals`` with a ``>=`` boundary (reached, not exceeded — H1): a HARD portfolio goal ``break``s the pass (``stopped_early``); a HARD per-project goal SKIPS that pid (its further runs are future passes, not more runs here); a SOFT goal flags ``stop_reason`` but continues. Because the ledger is static during the pass, a reached goal is observed on the first iteration. ``stop_reason`` surfaces the first goal event; a per-project skip is also observable as the pid's absence from ``runs``. ``concurrency`` (S3.3, D-D wave model) caps how many projects run at once. It is validated BEFORE anything loads — an unusable cap must surface as an error, never as a silently-empty pass. At the default ``1`` the pass is the sequential one described above. Error policy (S3.3, SC3) is COLLECT-AND-CONTINUE: a project that raises does not cancel its siblings and does not abort the pass. Its exception is recorded as a ``RunFailure`` in ``failures`` while every project that completed keeps its full ``RunResult``, so a portfolio of independent projects degrades one project at a time rather than all at once. This holds at every ``k``, including ``k=1``. The policy is carried by ``return_exceptions=True`` on the wave's ``gather``; ``asyncio.TaskGroup`` would cancel the siblings and is rejected for that reason. Note the pass still raises for errors that are NOT one project's failure — an unknown ``project_id`` and a non-positive ``concurrency`` are caller mistakes and fail fast. Goal-stop under waves (S3.3 reading of the Step-8 semantics above — a reading, NOT a redesign): the checks run at WAVE ASSEMBLY, per member, before the wave starts. A HARD per-project goal removes that pid from its wave, so an excluded project is never STARTED — it costs no model round-trip and leaves no verdict behind, which a post-hoc filter over completed runs could not achieve. A HARD portfolio goal stops the pass, and the wave already assembled still crosses its merge barrier before the loop exits, so a stop never strands verdicts outside the store. A SOFT goal flags ``stop_reason`` and continues. Membership is identical at every ``concurrency`` — but the reason is that the ledger is STATIC during a pass (C3: no realization happens on the run path, which ``test_concurrent_pass_does_not_write_on_the_run_path`` pins), so every check reads the same accumulated sum regardless of when it runs. Wave-assembly placement buys the never-started property, not determinism; determinism is the one-writer rule's. **The one semantic difference ``k > 1`` introduces, stated plainly (OQ1).** Every project in a wave reads the WAVE-START store, so a verdict captured by project A does NOT reach project B's hypothesis prompt when A and B share a wave — sequentially it would. This is a deliberate trade, not an oversight: the snapshot is what removes the append race, and restoring intra-wave visibility would restore exactly the completion-order dependence the barrier exists to eliminate (what B saw would depend on whether A happened to finish first). Learning therefore flows ACROSS wave boundaries, not within them, and ``concurrency`` is the knob that trades learning granularity for wall-clock — at ``k=1`` nothing changes, and at ``k=len(portfolio)`` the pass learns nothing from itself. On the SHIPPED reference fixture this difference is invisible in outcomes, because the Step-1 ExpeL fold is ``bundle_dir``-gated and no reference project sets ``bundle_dir`` — the chain is live for store CONTENT but inert for OUTCOMES. That is a property of the fixture, never a property of the design, so it is pinned on a bundle-backed pair where the fold does fire (``test_intra_wave_visibility_is_the_documented_semantic_difference``): same fixture, same sentinel, only the wave boundary moves. Store content stays identical across ``k`` — the difference is confined to what each project READ, never to what the pass produced or persisted. ``portfolio_meter`` (S3.4, F10) installs the GLOBAL token cap. Without it nothing changes: each run is bounded only by its own ``max_tokens``, so N projects can cost N times that with no ceiling over the pass. With it, one ``PortfolioMeter`` is shared by every run (each run's meter is BOUND to it, which is why ``meter_factory`` — whose meters are unbound — is refused alongside it: accepting both would run a pass that looks capped and is not), and the cap is enforced in three places that are deliberately different: - **At startup**, a remainder that cannot fund one run raises ``BudgetRefused`` — a pass that can afford zero projects is a caller mistake, not a result. - **At wave assembly**, a project that cannot be funded is NEVER STARTED, and the pass stops with ``budget_stop`` + ``stopped_early``, every completed run preserved. Never-started is the property that matters: an unfunded project that is merely interrupted mid-run has already cost model round-trips. Because every member of a wave is checked against the SAME pre-wave remainder, admission RESERVES each member's requirement as it goes — otherwise a wave of k would over-commit the cap by up to k runs. This makes membership identical at every ``concurrency``, matching the goal-stop's guarantee above. - **Before each chat call** (``BudgetMiddleware``'s pre-call guard), a call the remainder cannot pay for is refused rather than made. This is what bounds a run that was funded at admission but whose siblings drained the pass while it was in flight; such a run surfaces as a ``RunFailure``, not as overspend. Spend is carried ACROSS passes by seeding the meter from ``budget.read_spend`` and writing ``budget.write_spend`` afterwards. Those are the caller's calls, not this function's — the pass reads its ledger, it does not own the file (mirroring the Steg-7 role split).""" if concurrency < 1: raise ValueError( f"concurrency must be >= 1, got {concurrency}: a non-positive wave size would run no " "projects at all and read as an empty portfolio. Pass 1 for the sequential pass." ) if portfolio_meter is not None and meter_factory is not None: raise ValueError( "portfolio_meter and meter_factory are mutually exclusive: a meter_factory meter is " "not bound to the portfolio ledger, so the global cap would be silently unenforced" ) # Startup refusal (fail-fast, before anything loads): a pass that cannot fund its first run # must not begin. Raised, not returned — an empty PortfolioResult would read as "nothing to do". if portfolio_meter is not None and not portfolio_meter.can_fund_run(): raise BudgetRefused(portfolio_meter.remaining(), portfolio_meter.required_per_run) projects = {p.id: p for p in load_reference_projects()} ids = list(project_ids) if project_ids is not None else list(projects) store = store if store is not None else VerdictStore(verdicts=[]) # S3.1: the flag is FORWARDED per project rather than installed on the shared store here. The # store is threaded across every project in the pass (and may be caller-owned), so a pre-loop # assignment outlived the pass; forwarding keeps the opt-in scoped to each run's own retrievals. ledger = ledger if ledger is not None else SavingsLedger(entries=[]) goals = goals if goals is not None else GoalConfig() portfolio_baseline_ore = _baseline_ore(projects[p] for p in ids if p in projects) runs: list[RunResult] = [] failures: list[RunFailure] = [] stopped_early = False stop_reason: GoalReached | None = None budget_stop: BudgetStop | None = None for wave_ids in _waves(ids, concurrency): members: list[str] = [] # Tokens committed to members already admitted to THIS wave but that have not spent yet. # Every member reads the same pre-wave remainder, so without this a wave of k would admit # k projects off one project's worth of budget. reserved = 0 for pid in wave_ids: if pid not in projects: raise ValueError(f"unknown project_id: {pid!r}") project = projects[pid] if goals.portfolio is not None: observed = ledger.portfolio_total() limit = _goal_limit_if_reached(goals.portfolio, observed, portfolio_baseline_ore) if limit is not None: if goals.portfolio.mode == "hard": stopped_early = True stop_reason = GoalReached("portfolio", None, limit, observed) break if stop_reason is None: stop_reason = GoalReached("portfolio", None, limit, observed) # soft flag per_project_goal = goals.per_project.get(pid) if per_project_goal is not None: observed = ledger.per_project_total(pid) limit = _goal_limit_if_reached( per_project_goal, observed, _baseline_ore((project,)) ) if limit is not None: if stop_reason is None: stop_reason = GoalReached("project", pid, limit, observed) if per_project_goal.mode == "hard": continue # skip THIS pid; the rest of the pass proceeds # S3.4 funding check, placed AFTER the goal checks: a reached goal is success and owns # the stop when both apply, and a pid a hard per-project goal already skipped costs no # budget, so it must not consume a reservation. if portfolio_meter is not None and not portfolio_meter.can_fund_run(reserved=reserved): stopped_early = True budget_stop = BudgetStop( limit_tokens=portfolio_meter.budget.max_total_tokens, spent_tokens=portfolio_meter.spent, remaining_tokens=portfolio_meter.remaining(), required_tokens=portfolio_meter.required_per_run, ) break if portfolio_meter is not None: reserved += portfolio_meter.required_per_run members.append(pid) # Every project in the wave reads the SAME wave-start state and writes only its own copy, # so no two coroutines touch one list. The snapshot is what makes the barrier sufficient. snapshots = [(pid, _wave_snapshot(store)) for pid in members] # run_portfolio only drives full runs (never dry-run), so the return narrows to RunResult; # the cast keeps the widened run_project signature honest without an @overload duplication. # ``return_exceptions=True`` is the collect-and-continue seam (SC3): one project's # exception must not cancel its siblings. ``asyncio.TaskGroup`` is deliberately NOT used — # it cancels the remaining tasks on first exception, which is this policy's exact negation. wave_results = await asyncio.gather( *( run_project( pid, profile, docs_dir=projects[pid].docs_dir, verdict_input=projects[pid].verdict_input, bundle_dir=projects[pid].bundle_dir, verdict_dir=projects[pid].verdict_dir, dimension=dimension, store=snapshot, client_factory=client_factory, max_rounds=max_rounds, max_tokens=max_tokens, top_k=top_k, semantic_retrieval=semantic_retrieval, embedder=embedder, meter=_run_meter(meter_factory, portfolio_meter, max_rounds), ) for pid, snapshot in snapshots ), return_exceptions=True, ) # ``gather`` resolves in ARGUMENT order, not completion order, and waves follow # ``project_ids`` — so ``runs`` stays in caller order however the schedule interleaved, and # position is a sound key for pairing each result back to the pid that produced it. # # ``strict=True`` is FUTURE-PROOFING and is deliberately untested — measured, not assumed: # dropping it leaves the whole suite green, because ``gather`` is constructed from exactly # ``snapshots``, so the two lengths cannot diverge today. A test could only go red by # manufacturing a mismatch, which would exercise ``zip`` rather than this pass. It earns its # place against a later edit that filters or extends one sequence without the other — then a # silent truncation would mis-attribute every result after the gap, and this fails instead. for (pid, _snapshot), outcome in zip(snapshots, wave_results, strict=True): # ``BaseException``, NOT ``Exception``, and the width is load-bearing: a member raising # ``asyncio.CancelledError`` (a BaseException since 3.8) is COLLECTED by # ``return_exceptions=True`` and must land in ``failures``. Narrowed to ``Exception`` it # would fall to the ``cast`` below and put a live exception object into ``runs``, which # dies later in ``_aggregate`` — a crash pointing away from its cause. Gated by # ``tests/test_portfolio_failure_accounting_loadbearing.py``. if isinstance(outcome, BaseException): failures.append( RunFailure( project_id=pid, error=str(outcome), error_type=type(outcome).__name__ ) ) else: runs.append(cast(RunResult, outcome)) # ``snapshots`` is handed over UNFILTERED. ``_merge_wave`` consumes the ORDER this sequence # arrives in and nothing else, so that order is the contract. Measured honestly: filtering # the failed members out here would in fact be harmless, because filtering preserves # relative order — the variant is green. What is NOT harmless is anything that REORDERS # (or re-pairs) the sequence, and keeping the list whole is what leaves no place for that # mistake to be written. A failed project's snapshot holds no new verdicts, so it costs # nothing to pass through. _merge_wave(store, snapshots) if stopped_early: break base = _aggregate(tuple(runs), store) if stopped_early or stop_reason is not None or failures or budget_stop is not None: return replace( base, stopped_early=stopped_early, stop_reason=stop_reason, failures=tuple(failures), budget_stop=budget_stop, ) return base # The roles ``debate``/``generate`` ask the factory for. Fixed here so a malformed replies file is # caught at the door instead of mid-run. _SCRIPTED_ROLES = ("proposer", "checker") # The honesty banner for the scripted door. It is a REQUIREMENT, not decoration (målbilde §1): # a scripted run that reads like a model run is worse than having no offline mode at all, so this # prints on every scripted invocation and mirrors ``simulation.main``'s banner. _SCRIPTED_BANNER = ( "=" * 78 + "\nSCRIPTED OFFLINE RUN — every agent reply is read from your --scripted-replies file." + "\nNO MODEL WAS CALLED (ingen modellkall gjort). The context navigation, the debate" + "\nplumbing, the deterministic validator and the verdict are real; the agents' answers" + "\nare yours, not a model's. This proves the loop closes — not that an LLM would say this." + "\n" + "=" * 78 ) def _load_scripted_replies(path: str) -> dict[str, str]: """Load the caller's scripted answers, fail-fast. Every role the debate can ask for must be present AND a string: a missing role would otherwise surface as a ``KeyError`` deep inside ``scripted_factory``'s lookup, mid-run, long after the run appeared to start cleanly.""" try: raw = json.loads(Path(path).read_text(encoding="utf-8")) except FileNotFoundError as exc: raise ValueError(f"--scripted-replies file not found: {path}") from exc except json.JSONDecodeError as exc: raise ValueError(f"--scripted-replies is not valid JSON ({path}): {exc}") from exc if not isinstance(raw, dict): raise ValueError(f"--scripted-replies must be a JSON object of role -> reply ({path})") missing = [r for r in _SCRIPTED_ROLES if not isinstance(raw.get(r), str)] if missing: raise ValueError( f"--scripted-replies needs a string reply for each of {', '.join(_SCRIPTED_ROLES)}; " f"missing or non-string: {', '.join(missing)} ({path})" ) return {role: raw[role] for role in _SCRIPTED_ROLES} def main(argv: list[str] | None = None) -> int: """Single-command console entry: run the slice for one project against a docs folder.""" import argparse import asyncio import sys parser = argparse.ArgumentParser(description="portfolio-optimiser vertical slice") # project_id + --docs-dir are relaxed from required to a mode-conditional refusal (below): the # single-project path still requires both, but portfolio mode takes neither. The compensating # guard keeps the legacy contract failing loudly (rc 1 refusal) instead of via argparse exit 2. parser.add_argument("project_id", nargs="?", default=None) parser.add_argument("--profile", default="local") parser.add_argument("--docs-dir", default=None) parser.add_argument( "--bundle-dir", default=None, help="OKF bundle dir (enables the Step-1 fold)" ) parser.add_argument( "--verdict-dir", default=None, help="async verdict inbox: a folder of dropped expert verdicts, ingested before generation " "(the long loop — a verdict that landed after an earlier run is consumed by this run)", ) parser.add_argument( "--dimension-config", default=None, help="fail-fast dimension scope config (JSON): scopes the run to one cost axis; a " "missing or malformed file refuses the run (authoritative startup config, not a RAW inbox)", ) parser.add_argument( "--embedder-config", default=None, help='fail-fast embedder config (JSON, e.g. {"type": "fake"}): selects the embedder ' "used by --semantic-retrieval from a CLOSED registry. Not an import path — a config file " "can never name arbitrary code to load (a new embedder is added as a registry branch)", ) parser.add_argument( "--outbox-dir", default=None, help="RAW outbox dir for the run's proposal/outcome artefacts (REQUIRES --run-id). MUST " "differ from --verdict-dir: writing the outbox into a folder later read as an inbox " "re-ingests raw agent output past the Step-8 promotion gate (self-contamination) — " "documented, deliberately NOT CLI-enforced", ) parser.add_argument( "--run-id", default=None, help="stable run id for --outbox-dir artefacts (required when --outbox-dir is set; no " "wall-clock/uuid default — the outbox artefacts are byte-deterministic)", ) parser.add_argument( "--portfolio", action="store_true", help="portfolio mode: dispatch to run_portfolio over all reference projects (or the single " "given PROJECT_ID). Takes --goals/--ledger/--dimension-config; the single-project-only flags " "are refused in this mode (the two CLI modes are a documented partition)", ) parser.add_argument( "--goals", default=None, help="portfolio mode: goal config JSON (fail-fast) — the GoalReached stop is checked against " "the ledger before each project", ) parser.add_argument( "--ledger", default=None, help="portfolio mode: accumulated savings ledger JSON (fail-fast) read for the goal-stop " "(earlier out-of-band HITL realizations — never built during the pass)", ) parser.add_argument( "--semantic-retrieval", action="store_true", help="S3.1 opt-in scaling SEAM: blend a cosine term over the embedded feature triple with " "the structural score, so a prior verdict on a DIFFERENT cost-code set can outrank one that " "ties structurally. The shipped embedder is a semantics-free sha256 projection — this buys " "the extension point, not better retrieval; inject a real one with --embedder-config. " "Accepted in both modes, but in single-project mode it REQUIRES --bundle-dir and " "--verdict-dir (without them it cannot take effect, and is refused rather than ignored). " "OFF by default, and off means the structural ranking is unchanged", ) parser.add_argument("--decision", default="approved", choices=["approved", "rejected"]) parser.add_argument("--rationale", default="reviewed by expert") parser.add_argument( "--live-dry-run", action="store_true", help="offline drill: build contracts/clients/budget, STOP before the first model call", ) parser.add_argument( "--scripted-replies", default=None, metavar="FILE", help="offline WHOLE-LOOP run over your own bundle with ZERO model calls: FILE is JSON " '{"proposer": "", "checker": ""} and those fixed strings stand in for every ' "model answer. Unlike --live-dry-run (which stops before the first call) the complete loop " "runs — hypothesis, debate, deterministic validator, verdict. The answers are yours, not a " "model's, and the run says so on every invocation", ) parser.add_argument( "--report", action="store_true", help="S5.4 read-only value report: roll up the --ledger's realized savings (per-project + " "portfolio totals, flagged cross-dimension overlaps, per-entry provenance) to stdout. " "Mode-exclusive: only --ledger/--json are permitted alongside it; makes NO model calls", ) parser.add_argument( "--json", action="store_true", help="value report output form (requires --report): emit the roll-up as deterministic JSON " "instead of the human table", ) args = parser.parse_args(argv) # S5.4: read-only value-report dispatch — placed FIRST (right after parse_args, BEFORE the # mode-exclusivity block below) so it returns before any model/portfolio path can start and no # later branch can shadow it (the bare `--ledger`-outside-portfolio refusal at the elif below is # left UNCHANGED — a bare --ledger with no --report still flows there and refuses as before). if args.json and not args.report: # A stray --json is never silently ignored (honors S5.3's "refused, never ignored" partition). print("run refused: --json requires --report", file=sys.stderr) return 1 if args.report: # Mode-exclusivity as an ALLOWLIST (not a short blocklist): report mode permits ONLY --ledger # and --json; ANY other distinguishable mode/config flag is refused — else --report --goals # would silently drop --goals, whereas bare --goals is refused below (adding --report must not # suppress an existing refusal). --decision/--rationale are excluded: their non-None argparse # defaults are indistinguishable from an explicit value (exactly as the block below excludes # them); they are inert in report mode. report_forbidden = { "--portfolio": args.portfolio, "--live-dry-run": args.live_dry_run, "PROJECT_ID": args.project_id is not None, "--goals": args.goals is not None, "--docs-dir": args.docs_dir is not None, "--bundle-dir": args.bundle_dir is not None, "--verdict-dir": args.verdict_dir is not None, "--outbox-dir": args.outbox_dir is not None, "--run-id": args.run_id is not None, "--dimension-config": args.dimension_config is not None, "--semantic-retrieval": args.semantic_retrieval, "--embedder-config": args.embedder_config is not None, "--scripted-replies": args.scripted_replies is not None, } if any(report_forbidden.values()): print( "run report refused: mode-exclusive (only --ledger/--json permitted with --report)", file=sys.stderr, ) return 1 if not args.ledger: # Guards SavingsLedger.load(None) -> Path(None) TypeError (NOT in the load except tuple). print("run report refused: --report requires --ledger ", file=sys.stderr) return 1 try: # `report_ledger`, not `ledger`: the portfolio branch below binds `ledger` as # `SavingsLedger | None`, so reusing that name here (type `SavingsLedger`) collides on # mypy's function-scoped declared type. report_ledger = SavingsLedger.load(args.ledger) except (FileNotFoundError, ValidationError, ValueError) as exc: # A load failure must never masquerade as a real zero-savings result (SC5): stderr + rc 1, # no table. Only a successfully-loaded (possibly empty) ledger prints. print(f"run report refused: {exc}", file=sys.stderr) return 1 rep = build_value_report(report_ledger) # NB: `rep`, not `report` (`report` is bound below) print(dump_report_json(rep) if args.json else format_report_text(rep)) return 0 # Step 4: mode-exclusivity validation (structured refusal, NOT argparse.error — keeps the rc 1 # refusal contract). The two CLI modes are a documented partition: single-project-only flags are # refused in portfolio mode, and --goals/--ledger are refused outside it — never silently ignored. # --decision/--rationale are excluded: their non-None argparse defaults make an explicit value # indistinguishable from the default, so an honest refusal is unimplementable (they are inert in # portfolio mode; the README documents that). --dimension-config is valid in BOTH modes. if args.portfolio: single_only = { "--docs-dir": args.docs_dir, "--bundle-dir": args.bundle_dir, "--verdict-dir": args.verdict_dir, "--outbox-dir": args.outbox_dir, "--run-id": args.run_id, "--live-dry-run": args.live_dry_run, } offending = [name for name, value in single_only.items() if value] if offending: print( f"portfolio run refused: {', '.join(offending)} belong to single-project mode, " "not --portfolio (the two CLI modes are a documented partition)", file=sys.stderr, ) return 1 elif args.goals is not None or args.ledger is not None: print( "run refused: --goals/--ledger require --portfolio mode", file=sys.stderr, ) return 1 if args.portfolio: # Portfolio mode (Step 3): dispatch to the EXISTING run_portfolio via the fail-fast loaders # (run_portfolio itself is unchanged). Loader/ValueError failures surface through the same # structured-refusal contract as the single-project path (stderr + rc 1, no traceback). try: goals = load_goal_config(args.goals) if args.goals else None ledger = SavingsLedger.load(args.ledger) if args.ledger else None dimension = load_dimension(args.dimension_config) if args.dimension_config else None embedder = ( build_embedder(load_embedder_config(args.embedder_config)) if args.embedder_config else None ) project_ids = (args.project_id,) if args.project_id is not None else None portfolio_result = asyncio.run( run_portfolio( project_ids, args.profile, dimension=dimension, embedder=embedder, ledger=ledger, goals=goals, semantic_retrieval=args.semantic_retrieval, ) ) except (ValueError, FileNotFoundError, ValidationError) as exc: print(f"portfolio run refused: {exc}", file=sys.stderr) return 1 for r in portfolio_result.runs: print(f"{type(r.outcome).__name__}: verdict id={r.verdict.id}") if portfolio_result.stop_reason is not None: sr = portfolio_result.stop_reason print( f"goal reached: scope={sr.scope} project={sr.project_id or '-'} " f"observed_ore={sr.observed_ore} limit_ore={sr.limit_ore} " f"stopped_early={portfolio_result.stopped_early}" ) return 0 # Single-project mode requires PROJECT_ID + --docs-dir (compensating for the relaxed argparse # required/positional so the legacy contract keeps failing loudly via the refusal surface). if args.project_id is None or args.docs_dir is None: print( "run refused: single-project mode requires PROJECT_ID and --docs-dir " "(use --portfolio for portfolio mode)", file=sys.stderr, ) return 1 # --semantic-retrieval is refused, never silently ignored (the repo's flag contract). In # single-project mode it can only do observable work with BOTH of these: the Step-1 fold is # gated on ``bundle_dir``, and ``--verdict-dir`` is the only route by which ``main()`` can hand # ``run_project`` a non-empty store (``main()`` never passes ``store=``, and ``run_project`` # never seeds one). Without them the flag would rank nothing that reaches a prompt, and # ``RunResult.retrieved`` never leaves the process — ``main()`` prints one outcome line only. # # DELIBERATELY STATIC. There is no runtime "refuse if the store ends up empty" check: a # missing, empty or partially-skipped inbox is the Steg-7 tolerant-load contract, so refusing # there would fire on a legitimate first run. The refusal is therefore necessary, not # sufficient — it catches the configuration that CANNOT work, not every run that finds nothing. # # main() only. As a library API, ``run_project(semantic_retrieval=True, store=…)`` with a # caller-supplied store stays legitimate — that is the path the tests drive. Portfolio mode is # unaffected: ``run_portfolio`` always resolves a store and populates it by cross-project capture. if args.semantic_retrieval: required = {"--bundle-dir": args.bundle_dir, "--verdict-dir": args.verdict_dir} missing = [name for name, value in required.items() if not value] if missing: print( f"run refused: --semantic-retrieval requires {' and '.join(missing)} in " "single-project mode (the Step-1 fold is bundle-path-only, and --verdict-dir is " "the only route to a non-empty store)", file=sys.stderr, ) return 1 # The scripted door (offline WHOLE-loop run over the caller's own bundle). Resolved BEFORE the # dry-run branch so the two offline modes cannot both be honoured. scripted_client_factory: Callable[[str], BaseChatClient] | None = None if args.scripted_replies is not None: if args.live_dry_run: # Both are offline, and they contradict: --live-dry-run stops before the first model # call while --scripted-replies answers every one of them. Refuse rather than let one # silently win (S5.3's "refused, never ignored" partition). print( "run refused: --scripted-replies and --live-dry-run are both offline modes and " "contradict each other (dry-run stops before the first model call; scripted " "answers all of them) — pick one", file=sys.stderr, ) return 1 try: replies = _load_scripted_replies(args.scripted_replies) except (OSError, ValueError) as exc: print(f"run refused: {exc}", file=sys.stderr) return 1 # Imported HERE rather than at module scope: ``simulation`` imports ``run``, so a top-level # import would be circular. The scripted client already exists as MAF-side scaffolding — # this flag is a DOOR onto that one seam, never a second implementation of it. from portfolio_optimiser.simulation import scripted_factory scripted_client_factory = scripted_factory(replies, []) print(_SCRIPTED_BANNER) if args.live_dry_run: # S4.2 drill (comparison protocol §4 pkt 2/3): walk the offline path, STOP before the first # model call, print the run-config. A misconfigured profile (e.g. AZURE with a # REPLACE-WITH-* placeholder) makes resolve_model raise inside the eager factory build — # refuse cleanly (mirror preflight.main) instead of tracebacking; S4.1 is the config gate. try: report = asyncio.run( run_project( args.project_id, args.profile, docs_dir=args.docs_dir, bundle_dir=args.bundle_dir, verdict_dir=args.verdict_dir, dimension=( load_dimension(args.dimension_config) if args.dimension_config else None ), embedder=( build_embedder(load_embedder_config(args.embedder_config)) if args.embedder_config else None ), outbox_dir=args.outbox_dir, run_id=args.run_id, verdict_input={"decision": args.decision, "rationale": args.rationale}, live_dry_run=True, ) ) except (ValueError, FileNotFoundError, ValidationError) as exc: # Structured refusal (rc 1, no traceback) for ANY offline-path ValueError. The # azure-preflight remediation is only meaningful for the AZURE config gate (S4.1), so # scope it to that profile — a LOCAL-profile ValueError (unknown project_id, empty # docs_dir, bundle mismatch) must not carry an irrelevant azure hint. msg = f"live-dry-run refused: {exc}" if args.profile == "azure": msg += ( "\nkjør 'python -m portfolio_optimiser.preflight --profile azure' først " "(S4.1 offline config-gate)" ) print(msg, file=sys.stderr) return 1 assert isinstance(report, DryRunReport) # live_dry_run=True always returns a DryRunReport print( f"{args.project_id}: LIVE-DRY-RUN OK (profile={report.profile}, " f"models={report.resolved_models}, max_rounds={report.max_rounds}, " f"max_tokens={report.max_tokens}, top_k={report.top_k}) — " "ingen modellkall gjort (stoppet før første debate.run)" ) return 0 try: result = cast( RunResult, asyncio.run( run_project( args.project_id, args.profile, docs_dir=args.docs_dir, bundle_dir=args.bundle_dir, verdict_dir=args.verdict_dir, dimension=( load_dimension(args.dimension_config) if args.dimension_config else None ), embedder=( build_embedder(load_embedder_config(args.embedder_config)) if args.embedder_config else None ), outbox_dir=args.outbox_dir, run_id=args.run_id, verdict_input={"decision": args.decision, "rationale": args.rationale}, semantic_retrieval=args.semantic_retrieval, client_factory=scripted_client_factory, ) ), ) except (ValueError, FileNotFoundError, ValidationError) as exc: # Structured refusal (rc 1, no traceback) for the full-run path: run_project's fail-fast # loaders (contracts, load_dimension, outbox run_id guard) surface here as one clean line. print(f"run refused: {exc}", file=sys.stderr) return 1 kind = type(result.outcome).__name__ print(f"{args.project_id}: {kind} (verdict id={result.verdict.id}, decision={args.decision})") return 0 if __name__ == "__main__": # pragma: no cover - console entry raise SystemExit(main())