feat(fase1): hard/soft goal-stop in run_portfolio on accumulated ledger (F1)

src/portfolio_optimiser/run.py

@@ -26,14 +26,16 @@ durable learned verdict captured out-of-band in the VerdictStore (D7-portable).
 from __future__ import annotations
 
 from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
-from typing import Any
+from decimal import ROUND_HALF_UP, Decimal
+from typing import Any, Literal
 
 from agent_framework import BaseChatClient, SessionContext
 
 from portfolio_optimiser.backends import Profile, get_backend, resolve_model
 from portfolio_optimiser.budget import Budget, BudgetMiddleware, TokenMeter
-from portfolio_optimiser.contracts import load_contracts
+from portfolio_optimiser.contracts import GoalConfig, GoalContract, load_contracts
+from portfolio_optimiser.ledger import SavingsLedger
 from portfolio_optimiser.datasource import (
     bundle_citations,
     chunk_dict_to_citation,
@@ -77,6 +79,20 @@ class RunResult:
     checker_verdict: str = "absent"
 
 
+@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 PortfolioResult:
     """The outcome of a sequential fan-out over N projects (SC2).
@@ -86,7 +102,8 @@ class PortfolioResult:
     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."""
+    provenance token usage. ``stopped_early`` / ``stop_reason`` record a Step-8 goal-stop: they
+    default so the frozen aggregate and every existing constructor call are unaffected."""
 
     runs: tuple[RunResult, ...]
     store: VerdictStore
@@ -94,6 +111,8 @@ class PortfolioResult:
     rejected_count: int
     sum_claimed_saving_nok: float
     sum_token_usage: int
+    stopped_early: bool = False
+    stop_reason: GoalReached | None = None
 
 
 def _authored_texts(result: Any, name: str) -> list[str]:
@@ -389,11 +408,31 @@ def _aggregate(runs: tuple[RunResult, ...], store: VerdictStore) -> PortfolioRes
     )
 
 
+def _to_ore(nok: float) -> int:
+    """NOK float -> integer øre, deterministically (Decimal, mirrors ``ledger.realize``)."""
+    return int((Decimal(str(nok)) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
+
+
+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:
+        threshold = int(goal.percent / 100 * baseline_ore)
+        if observed_ore >= threshold:
+            return threshold
+    return None
+
+
 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,
@@ -407,15 +446,53 @@ async def run_portfolio(
     across every run, so a verdict on project k informs the ExpeL retrieval of project k+1 (the
     cross-project learning loop). ``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)."""
+    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``."""
     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=[])
+    ledger = ledger if ledger is not None else SavingsLedger(entries=[])
+    goals = goals if goals is not None else GoalConfig()
+    portfolio_baseline_ore = _to_ore(sum(projects[p].total_cost for p in ids if p in projects))
+
     runs: list[RunResult] = []
+    stopped_early = False
+    stop_reason: GoalReached | None = None
     for pid in 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, _to_ore(project.total_cost))
+            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
+
         result = await run_project(
             pid,
             profile,
@@ -430,7 +507,11 @@ async def run_portfolio(
             meter=meter_factory() if meter_factory is not None else None,
         )
         runs.append(result)
-    return _aggregate(tuple(runs), store)
+
+    base = _aggregate(tuple(runs), store)
+    if stopped_early or stop_reason is not None:
+        return replace(base, stopped_early=stopped_early, stop_reason=stop_reason)
+    return base
 
 
 def main(argv: list[str] | None = None) -> int:

tests/test_ledger.py

@@ -12,7 +12,9 @@ from __future__ import annotations
 import pytest
 from pydantic import ValidationError
 
+from portfolio_optimiser.contracts import GoalConfig, GoalContract
 from portfolio_optimiser.ledger import LedgerEntry, SavingsLedger, stamp
+from portfolio_optimiser.run import run_portfolio
 
 _TS = "2026-07-06T00:00:00Z"
 
@@ -99,3 +101,115 @@ def test_save_is_byte_identical_regardless_of_order(tmp_path) -> None:
     # Round-trips: load() reconstructs an equivalent ledger with the same totals.
     reloaded = SavingsLedger.load(str(pa))
     assert reloaded.portfolio_total() == 7500
+
+
+# --- Step 8: hard/soft goal-stop in run_portfolio on the accumulated ledger (SC6/SC8) ------------
+
+_PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"]
+
+
+def _prefilled(project_id: str, amount_ore: int) -> SavingsLedger:
+    """A ledger prefilled with ONE realized entry (representing an EARLIER, out-of-band HITL
+    realization — the accumulated sum the goal-stop reads before this pass)."""
+    led = SavingsLedger()
+    led.add_realized(
+        LedgerEntry(
+            project_id=project_id,
+            dimension="energi",
+            candidate_identity="prior-hitl",
+            amount_ore=amount_ore,
+            verdict_id="v-prior",
+            provenance=stamp(approver="ekspert", experiment="earlier", timestamp=_TS),
+        )
+    )
+    return led
+
+
+async def test_portfolio_hard_goal_stops_the_whole_pass(make_portfolio_client_factory) -> None:
+    ledger = _prefilled("FV42-GSV-E1", 100)  # portfolio_total == 100
+    goals = GoalConfig(portfolio=GoalContract(absolute_ore=100))  # met at 100 (>=)
+    result = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=ledger,
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert result.stopped_early is True
+    assert result.runs == ()  # goal already reached before pid 0 -> nothing runs
+    assert result.stop_reason is not None
+    assert result.stop_reason.scope == "portfolio"
+    assert result.stop_reason.observed_ore == 100
+    assert result.stop_reason.limit_ore == 100
+
+
+async def test_boundary_exact_equal_stops(make_portfolio_client_factory) -> None:
+    """>= boundary: accumulated EXACTLY equal to the goal stops (reached, not strictly exceeded)."""
+    ledger = _prefilled("FV42-GSV-E1", 500)
+    goals = GoalConfig(portfolio=GoalContract(absolute_ore=500))
+    result = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=ledger,
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert result.stopped_early is True
+
+
+async def test_per_project_hard_goal_skips_only_that_pid(make_portfolio_client_factory) -> None:
+    ledger = _prefilled("FV42-GSV-E1", 1000)
+    goals = GoalConfig(per_project={"FV42-GSV-E1": GoalContract(absolute_ore=1000)})
+    result = await run_portfolio(
+        ["FV42-GSV-E1", "RV13-RAS-TP"],
+        "local",
+        ledger=ledger,
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    ran = [r.outcome.proposal.project_id for r in result.runs]
+    assert "FV42-GSV-E1" not in ran  # its goal is reached -> skipped
+    assert ran == ["RV13-RAS-TP"]  # the rest of the pass proceeds
+    assert result.stopped_early is False  # a per-project skip is NOT a pass-stop
+
+
+async def test_soft_goal_flags_but_continues(make_portfolio_client_factory) -> None:
+    ledger = _prefilled("FV42-GSV-E1", 100)
+    goals = GoalConfig(portfolio=GoalContract(absolute_ore=100, mode="soft"))
+    result = await run_portfolio(
+        ["FV42-GSV-E1", "RV13-RAS-TP"],
+        "local",
+        ledger=ledger,
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert result.stopped_early is False  # soft: does not stop
+    assert len(result.runs) == 2  # all ran
+    assert result.stop_reason is not None  # but the goal-reached flag IS surfaced
+
+
+async def test_stop_decision_is_deterministic(make_portfolio_client_factory) -> None:
+    """SC8: the same ledger + goals twice -> IDENTICAL stop decision (stopped_early + stop_reason)."""
+    goals = GoalConfig(portfolio=GoalContract(absolute_ore=100))
+    r1 = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=_prefilled("FV42-GSV-E1", 100),
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    r2 = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=_prefilled("FV42-GSV-E1", 100),
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert r1.stopped_early == r2.stopped_early
+    assert r1.stop_reason == r2.stop_reason  # frozen-dataclass equality: identical decision

tests/test_ledger_loadbearing.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import pytest
 
+from portfolio_optimiser.contracts import GoalConfig, GoalContract
 from portfolio_optimiser.ledger import (
     LedgerEntry,
     RealizationRefused,
@@ -25,6 +26,7 @@ from portfolio_optimiser.ledger import (
     realize,
     stamp,
 )
+from portfolio_optimiser.run import run_portfolio
 from portfolio_optimiser.verdicts import ProposalFeatures, Verdict
 
 _TS = "2026-07-06T00:00:00Z"
@@ -180,3 +182,53 @@ def test_realize_ore_conversion_is_exact() -> None:
         timestamp=_TS,
     )
     assert entry.amount_ore == 1234567
+
+
+# --- Step 8: goal-stop is load-bearing (SC6) -----------------------------------------------------
+
+_PORTFOLIO_IDS = ["FV42-GSV-E1", "RV13-RAS-TP", "BRU-LAKS-REHAB"]
+
+
+def _portfolio_ledger(amount_ore: int) -> SavingsLedger:
+    led = SavingsLedger()
+    led.add_realized(
+        LedgerEntry(
+            project_id="FV42-GSV-E1",
+            dimension="energi",
+            candidate_identity="prior-hitl",
+            amount_ore=amount_ore,
+            verdict_id="v-prior",
+            provenance=stamp(approver="ekspert", experiment="earlier", timestamp=_TS),
+        )
+    )
+    return led
+
+
+async def test_goal_stop_is_load_bearing(make_portfolio_client_factory) -> None:
+    """LOAD-BEARING (SC6): a reached HARD portfolio goal stops the pass (no project runs); a
+    below-goal ledger runs the FULL pass. RED if the goal-stop check is detached in run_portfolio
+    (a reached goal then runs past the target). The control (below goal -> full pass) proves the
+    stop is CAUSED by the goal being reached, not by the fixture."""
+    goals = GoalConfig(portfolio=GoalContract(absolute_ore=1000))
+
+    stopped = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=_portfolio_ledger(1000),  # == goal (>=)
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert stopped.stopped_early is True
+    assert stopped.runs == ()  # detach -> this becomes 3 (runs past the reached goal)
+
+    below = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        ledger=_portfolio_ledger(999),  # below the goal
+        goals=goals,
+        client_factory=make_portfolio_client_factory({}),
+        max_rounds=1,
+    )
+    assert below.stopped_early is False  # control: goal not reached -> full pass
+    assert len(below.runs) == 3

tests/test_portfolio.py

@@ -74,6 +74,23 @@ async def test_a_fanout_returns_one_runresult_per_project(
     assert all(r.provenance.token_usage > 0 for r in result.runs)
 
 
+async def test_a1_default_no_goal_leaves_stop_fields_unset(
+    make_portfolio_client_factory, fresh_store
+) -> None:
+    """Step 8 regression: with no ``goals``/``ledger`` the pass is unchanged — ``stopped_early`` is
+    False and ``stop_reason`` is None, and every project still runs (the new frozen fields default,
+    so the default path is byte-for-byte the prior behavior)."""
+    result = await run_portfolio(
+        _PORTFOLIO_IDS,
+        "local",
+        store=fresh_store,
+        client_factory=make_portfolio_client_factory(REPLIES),
+    )
+    assert result.stopped_early is False
+    assert result.stop_reason is None
+    assert len(result.runs) == 3
+
+
 async def test_a2_unknown_project_id_raises(make_portfolio_client_factory, fresh_store) -> None:
     """The unknown-id error path: an id absent from the loaded portfolio raises ValueError."""
     with pytest.raises(ValueError):