feat(fase5): add the long/async verdict file inbox (Steg 7 resumable feedback)
The short loop captured the expert verdict inline into an in-memory store, so a verdict arriving days/weeks later in a separate run could not influence any future hypothesis (målbilde §5 row 7). Steg 7 adds the long timescale: run_project gains an opt-in verdict_dir async inbox that load_verdicts_from_dir -> store.add MERGES into the store BEFORE the Step-1 ExpeL fold, so a verdict dropped after an earlier run reaches a separate, later run's hypothesis — fully resumable across runs separated in time. - verdicts.py: verdict_to_dict / verdict_from_dict (id read verbatim, never re-minted), write_verdict (public authoring primitive, NOT wired into run_project — system reads the folder, expert/persona writes it, §3 role split), tolerant load_verdicts_from_dir (missing/foreign/half-written files skipped, not raised — RAW layer per §10 R2), VerdictStore.from_dir. - run.py: verdict_dir kwarg; ingest-merge block after load_contracts (merge not replace keeps run_portfolio's cross-project threading; store.add idempotent on content-hash id; no change to the fold). CLI --bundle-dir/--verdict-dir thread the long loop to the console entry. No auto-persist of the run's own captured verdict (outbox/Steg 8). - Load-bearing PAIR (test_step7_async_loop_loadbearing.py): a verdict dropped after run A must reach run B's prompt (run B uses a FRESH store -> the transfer is the file loop, not in-memory carryover); empty-inbox control proves causality. Marker = a realization value absent from the bundle (not the seed's 0.82). Proven RED on ingest detach. Suite 138 -> 140 passed, 4 skipped; mypy + ruff check clean. Målbilde treated as frozen (no §3/§5/§7 edit). Step 8 (gated wiki promotion) remains. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MHR8iKxJRxDiDfNw8HZmWE
This commit is contained in:
parent
f7c81b45ec
commit
e2861cac0c
5 changed files with 297 additions and 6 deletions
|
|
@ -53,6 +53,7 @@ from portfolio_optimiser.verdicts import (
|
|||
VerdictStore,
|
||||
bundle_candidate_features,
|
||||
capture_verdict,
|
||||
load_verdicts_from_dir,
|
||||
)
|
||||
from portfolio_optimiser.workflow import fresh_workflow
|
||||
|
||||
|
|
@ -198,6 +199,7 @@ async def run_project(
|
|||
verdict_input: dict[str, str],
|
||||
bundle_dir: str | None = None,
|
||||
store: VerdictStore | None = None,
|
||||
verdict_dir: str | None = None,
|
||||
client_factory: Callable[[str], BaseChatClient] | None = None,
|
||||
max_rounds: int = 3,
|
||||
max_tokens: int = 100_000,
|
||||
|
|
@ -210,9 +212,13 @@ async def run_project(
|
|||
(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). Raises
|
||||
``pydantic.ValidationError`` on a bad contract and ``BudgetExceeded`` when the token/round cap
|
||||
is crossed."""
|
||||
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). Raises ``pydantic.ValidationError`` on a bad
|
||||
contract and ``BudgetExceeded`` when the token/round cap is crossed."""
|
||||
# 1. Fail-fast: validate ALL contracts (incl. the verdict-feedback shape) before any client.
|
||||
load_contracts(
|
||||
{"docs_dir": docs_dir, "top_k": top_k},
|
||||
|
|
@ -220,6 +226,15 @@ async def run_project(
|
|||
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``
|
||||
|
|
@ -406,6 +421,15 @@ def main(argv: list[str] | None = None) -> int:
|
|||
parser.add_argument("project_id")
|
||||
parser.add_argument("--profile", default="local")
|
||||
parser.add_argument("--docs-dir", required=True)
|
||||
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("--decision", default="approved", choices=["approved", "rejected"])
|
||||
parser.add_argument("--rationale", default="reviewed by expert")
|
||||
args = parser.parse_args(argv)
|
||||
|
|
@ -415,6 +439,8 @@ def main(argv: list[str] | None = None) -> int:
|
|||
args.project_id,
|
||||
args.profile,
|
||||
docs_dir=args.docs_dir,
|
||||
bundle_dir=args.bundle_dir,
|
||||
verdict_dir=args.verdict_dir,
|
||||
verdict_input={"decision": args.decision, "rationale": args.rationale},
|
||||
)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from __future__ import annotations
|
|||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import ContextProvider, SessionContext
|
||||
|
|
@ -104,12 +105,111 @@ def capture_verdict(features: ProposalFeatures, decision: str, rationale: str) -
|
|||
)
|
||||
|
||||
|
||||
# --- Async file inbox (Fase 5, Steg 7): plain-JSON verdict serialization + tolerant folder load ---
|
||||
# The raw output layer is a plain JSON store (R2), NOT OKF — OKF is reserved for the promoted
|
||||
# (Step-8) layer. One verdict per file lets an expert/persona drop verdicts into a folder out of
|
||||
# band, and a separate later run pick them up (målbilde §3 long loop).
|
||||
|
||||
_REQUIRED_VERDICT_KEYS = {"id", "decision", "rationale", "proposal_features"}
|
||||
|
||||
|
||||
def verdict_to_dict(verdict: Verdict) -> dict[str, Any]:
|
||||
"""Serialize a ``Verdict`` to a JSON-ready dict. ``affected_codes`` (a frozenset) is emitted as
|
||||
a SORTED list — matching ``_mint_id``'s canonical form — so the round-trip is lossless and the
|
||||
on-disk form is deterministic."""
|
||||
f = verdict.proposal_features
|
||||
return {
|
||||
"id": verdict.id,
|
||||
"decision": verdict.decision,
|
||||
"rationale": verdict.rationale,
|
||||
"proposal_features": {
|
||||
"affected_codes": sorted(f.affected_codes),
|
||||
"measure_type": f.measure_type,
|
||||
"claimed_saving_nok": f.claimed_saving_nok,
|
||||
"description": f.description,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def verdict_from_dict(data: dict[str, Any]) -> Verdict:
|
||||
"""Reconstruct a ``Verdict`` from its serialized dict. The ``id`` is read VERBATIM — never
|
||||
re-minted — so a verdict authored elsewhere keeps its identity and the int/float magnitudes in
|
||||
its features survive the JSON round-trip untouched (re-minting would re-hash and could diverge,
|
||||
since ``_mint_id`` hashes the raw value, e.g. ``30000`` vs ``30000.0``)."""
|
||||
pf = data["proposal_features"]
|
||||
return Verdict(
|
||||
id=data["id"],
|
||||
proposal_features=ProposalFeatures(
|
||||
affected_codes=frozenset(pf["affected_codes"]),
|
||||
measure_type=pf["measure_type"],
|
||||
claimed_saving_nok=pf["claimed_saving_nok"],
|
||||
description=pf.get("description", ""),
|
||||
),
|
||||
decision=data["decision"],
|
||||
rationale=data["rationale"],
|
||||
)
|
||||
|
||||
|
||||
def write_verdict(directory: str, verdict: Verdict) -> Path:
|
||||
"""Write one verdict as ``{id}.json`` into ``directory`` (created if needed) and return the path.
|
||||
|
||||
This is the public authoring primitive the expert/persona (sim) or human (prod) uses to drop a
|
||||
verdict into the async inbox via the SAME folder interface — and the writer a future Step-8
|
||||
promotion would reuse. It is deliberately NOT called inside ``run_project``: the system READS the
|
||||
inbox, it does not write to it (målbilde §3 role split).
|
||||
|
||||
Limitation: ``id`` is a content hash of the FEATURES only (``_mint_id``), so two verdicts with
|
||||
identical features but different decisions share a filename — last write wins on disk (mirroring
|
||||
``VerdictStore.add``'s first-wins in memory). Acceptable for the raw MVP layer."""
|
||||
path = Path(directory)
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
target = path / f"{verdict.id}.json"
|
||||
target.write_text(
|
||||
json.dumps(verdict_to_dict(verdict), sort_keys=True, indent=2), encoding="utf-8"
|
||||
)
|
||||
return target
|
||||
|
||||
|
||||
def load_verdicts_from_dir(directory: str) -> list[Verdict]:
|
||||
"""Load every well-formed verdict JSON file from an async inbox folder, TOLERANTLY (OKF §4
|
||||
spirit): a missing folder yields ``[]``; files that are not ``.json``, fail to parse, or lack a
|
||||
required key are SKIPPED, never raised. The folder is written out of band by an external party,
|
||||
so half-written or foreign files (e.g. a bundle's ``golden.json``) are realistic — this is the
|
||||
raw layer, not required input (contrast ``okf.load_ir_projection``'s fail-fast). Order is
|
||||
deterministic (sorted by filename)."""
|
||||
path = Path(directory)
|
||||
if not path.is_dir():
|
||||
return []
|
||||
verdicts: list[Verdict] = []
|
||||
for file in sorted(path.glob("*.json")):
|
||||
try:
|
||||
data = json.loads(file.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
continue
|
||||
if not isinstance(data, dict) or not _REQUIRED_VERDICT_KEYS <= data.keys():
|
||||
continue
|
||||
try:
|
||||
verdicts.append(verdict_from_dict(data))
|
||||
except (KeyError, TypeError):
|
||||
continue
|
||||
return verdicts
|
||||
|
||||
|
||||
@dataclass
|
||||
class VerdictStore:
|
||||
"""Minimal in-memory store of historical verdicts (MVP — no file persistence)."""
|
||||
"""Minimal store of historical verdicts. In-memory by default; ``from_dir`` / the
|
||||
``load_verdicts_from_dir`` merge primitive back it with the async file inbox (Steg 7)."""
|
||||
|
||||
verdicts: list[Verdict]
|
||||
|
||||
@classmethod
|
||||
def from_dir(cls, directory: str) -> VerdictStore:
|
||||
"""Build a store from an async inbox folder (convenience constructor). ``run_project`` uses
|
||||
the merge primitive (``add`` per loaded verdict) instead, to preserve a passed store's
|
||||
existing verdicts (the cross-project threading in ``run_portfolio``); this is for callers
|
||||
that want a fresh store straight from a folder."""
|
||||
return cls(verdicts=load_verdicts_from_dir(directory))
|
||||
|
||||
def retrieve(self, query: ProposalFeatures, k: int) -> list[Verdict]:
|
||||
"""Return the top-``k`` verdicts by structural similarity. Deterministic: ties break
|
||||
by verdict id, so ordering is stable across runs."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue