diff --git a/src/quantifact/__init__.py b/src/quantifact/__init__.py
index 7a49214..f8158ea 100644
--- a/src/quantifact/__init__.py
+++ b/src/quantifact/__init__.py
@@ -19,10 +19,12 @@
from .contracts.verdict import TaskUnfixable, Verdict
from .data.adapters.base import Adapter
from .data.adapters.demo_synthetic import DemoSyntheticAdapter
+from .data.conformance import AdapterConformanceReport, check_adapter
from .data.registry import SeriesMeta, SeriesStore
from .data.search import SeriesSearch
+from .evidence import ResearchEvidencePackage
from .harness.cache import ValueCache
-from .harness.execute import ExecutionHarness
+from .harness.execute import ExecutionHarness, ProcessExecutionHarness
from .learn.benchmarks import Benchmark, BenchmarkSuite
from .learn.lessons import Lesson, LessonRepo
from .learn.teach import teach
@@ -36,13 +38,14 @@
ResearchDesign,
Task,
)
-from .planner import RulePlanner
+from .planner import RulePlanner, UnsupportedQuestionError
__all__ = [
"ANALYST",
"AlternativeExplanation",
"PM",
"Adapter",
+ "AdapterConformanceReport",
"AnalysisPlan",
"Artifacts",
"Benchmark",
@@ -55,8 +58,10 @@
"LessonRepo",
"PlanCompiler",
"PlanError",
+ "ProcessExecutionHarness",
"Quantifact",
"ReferenceCodegen",
+ "ResearchEvidencePackage",
"ResearchClaim",
"ResearchDesign",
"RulePlanner",
@@ -66,9 +71,11 @@
"Task",
"TaskUnfixable",
"User",
+ "UnsupportedQuestionError",
"ValueCache",
"Verdict",
"generate_all",
+ "check_adapter",
"teach",
]
-__version__ = "0.2.0"
+__version__ = "0.3.0a1"
diff --git a/src/quantifact/agent.py b/src/quantifact/agent.py
index c502b27..fd11adf 100644
--- a/src/quantifact/agent.py
+++ b/src/quantifact/agent.py
@@ -28,12 +28,19 @@
from .codegen.base import CodegenBackend, generate_all, schemas_of
from .codegen.reference import ReferenceCodegen
from .contracts.layers import validate_result, validate_static
+from .contracts.methods import validate_method_evidence
from .contracts.reasoning import validate_claim_evidence
from .contracts.verdict import TaskUnfixable, Verdict
from .data.adapters.demo_synthetic import DemoSyntheticAdapter
from .data.registry import SeriesMeta
+from .evidence import ResearchEvidencePackage, build_evidence_package
from .harness.cache import ValueCache
-from .harness.execute import ExecutionHarness, RunResult, TaskExecutionError
+from .harness.execute import (
+ ExecutionHarness,
+ ProcessExecutionHarness,
+ RunResult,
+ TaskExecutionError,
+)
from .learn.lessons import LessonRepo
from .learn.workflows import WorkflowRepo
from .plan.compile import PlanCompiler
@@ -69,6 +76,8 @@ class Artifacts:
fix_rounds: int = 0
planning_trace: dict[str, Any] = field(default_factory=dict)
repair_trace: list[dict[str, Any]] = field(default_factory=list)
+ evidence: ResearchEvidencePackage | None = None
+ evidence_path: Path | None = None
@property
def layers(self) -> list[list[str]]:
@@ -156,13 +165,23 @@ def __init__(
debugger: Any | None = None,
semantic_validator: Any | None = None,
planner_backend: Any | None = None,
+ execution_mode: str = "in_process",
+ task_timeout_seconds: float = 30.0,
):
self.ws = Path(workspace)
self.ws.mkdir(parents=True, exist_ok=True)
self.user = user
self.adapter = adapter or DemoSyntheticAdapter(self.ws / "store")
self.cache = ValueCache(self.ws / "cache", enabled=cache_enabled)
- self.harness = ExecutionHarness(self.adapter, self.cache)
+ if execution_mode == "in_process":
+ self.harness = ExecutionHarness(self.adapter, self.cache)
+ elif execution_mode == "process":
+ self.harness = ProcessExecutionHarness(
+ self.adapter, self.cache, timeout_seconds=task_timeout_seconds
+ )
+ else:
+ raise ValueError("execution_mode must be 'in_process' or 'process'")
+ self.execution_mode = execution_mode
self.backend = backend or ReferenceCodegen()
self.lessons = LessonRepo(self.ws / "context" / "lessons")
self.workflows = WorkflowRepo(self.ws / "context" / "workflows")
@@ -218,6 +237,7 @@ def analyse(
out: str | Path | None = None,
max_fix_rounds: int = 2,
writeback: bool = True,
+ evidence_out: str | Path | None = None,
on_stage: Callable[[str, float], None] | None = None,
) -> Artifacts:
timings: dict[str, float] = {}
@@ -295,6 +315,14 @@ def execute() -> RunResult:
if failed_reasoning:
raise TaskUnfixable(failed_reasoning[0])
+ method_verdicts = stage(
+ "method_contracts", lambda: validate_method_evidence(plan, result.frames)
+ )
+ verdicts += method_verdicts
+ failed_methods = [v for v in method_verdicts if not v.ok]
+ if failed_methods:
+ raise TaskUnfixable(failed_methods[0])
+
frame_verdicts = stage(
"validate",
lambda: [
@@ -389,7 +417,7 @@ def execute() -> RunResult:
),
)
- return Artifacts(
+ artifacts = Artifacts(
plan=plan,
codes=codes,
result=result,
@@ -402,6 +430,33 @@ def execute() -> RunResult:
planning_trace=planning_trace,
repair_trace=repair_trace,
)
+ package = stage(
+ "evidence",
+ lambda: build_evidence_package(
+ plan=plan,
+ codes=codes,
+ result=result,
+ findings=findings,
+ verdicts=verdicts,
+ timings=timings,
+ planning_trace=planning_trace,
+ repair_trace=repair_trace,
+ adapter=self.adapter,
+ backend=self.backend.name,
+ user=self.user.name,
+ report_path=report_path,
+ execution_mode=self.execution_mode,
+ ),
+ )
+ package_path = None
+ target = evidence_out
+ if target is None and out is not None:
+ target = Path(out).with_suffix(".evidence.json")
+ if target is not None:
+ package_path = package.save(target)
+ artifacts.evidence = package
+ artifacts.evidence_path = package_path
+ return artifacts
# ------------------------------------------------------------ fix loops
def _repair_static(
diff --git a/src/quantifact/cli.py b/src/quantifact/cli.py
index 6c9e189..d488811 100644
--- a/src/quantifact/cli.py
+++ b/src/quantifact/cli.py
@@ -10,6 +10,8 @@
qf evals run the benchmark suite
qf bench run the performance benchmarks
qf audit evidence-backed PAT maturity audit
+qf verify
verify a research evidence package offline
+qf adapter-check run the point-in-time adapter conformance suite
"""
from __future__ import annotations
@@ -80,6 +82,8 @@ def cmd_ask(args) -> int:
debugger=debugger,
semantic_validator=semantic,
planner_backend=_planner_backend(args),
+ execution_mode=args.execution,
+ task_timeout_seconds=args.task_timeout,
)
answers = json.loads(args.answers) if args.answers else {}
if args.as_of:
@@ -98,6 +102,7 @@ def stage(name: str, secs: float) -> None:
out=out,
on_stage=stage,
max_fix_rounds=args.fix_rounds,
+ evidence_out=args.evidence,
)
print(f"\nas_of {art.plan.as_of} (nothing published later was read)")
print(f"plan {len(art.plan.tasks)} tasks in {len(art.layers)} layers")
@@ -117,6 +122,8 @@ def stage(name: str, secs: float) -> None:
if art.fix_rounds:
print(f"repairs {art.fix_rounds} round(s)")
print(f"report {art.report_path}")
+ if art.evidence_path:
+ print(f"evidence {art.evidence_path} ({art.evidence.package_id[:12]})")
if args.receipt:
receipt = art.receipt(backend=qf.backend.name, user=qf.user.name)
if backend is not None:
@@ -280,9 +287,10 @@ def cmd_evals(args) -> int:
ws = Path(args.workspace)
qf = Quantifact(ws, _user(args.user))
suite = BenchmarkSuite(Path(args.dir) if args.dir else ws / "benchmarks")
- results = suite.run_all(
+ report = suite.report(
qf.adapter, LessonRepo(ws / "context" / "lessons").all(), qf.user.entitlements
)
+ results = report.results
if not results:
print("no benchmarks yet — run `qf teach` first")
return 0
@@ -293,6 +301,12 @@ def cmd_evals(args) -> int:
print(f" {f}")
failed += not r.passed
print(f"\n{len(results) - failed}/{len(results)} passing")
+ for family, counts in report.slices("family").items():
+ print(f" family {family:<24} {counts['passed']}/{counts['total']}")
+ print(f"silent critical failures: {report.silent_critical_failures}")
+ if args.json:
+ Path(args.json).write_text(json.dumps(report.to_dict(), indent=2))
+ print(f"wrote {args.json}")
return 1 if failed else 0
@@ -321,6 +335,43 @@ def cmd_audit(args) -> int:
return 1 if args.strict and report.level != "PAT-level evidence" else 0
+def cmd_verify(args) -> int:
+ """Verify internal package integrity; this is not publisher authentication."""
+ from .evidence import ResearchEvidencePackage
+
+ package = ResearchEvidencePackage.load(args.package)
+ problems = package.verify()
+ if problems:
+ print("INVALID")
+ for problem in problems:
+ print(f" - {problem}")
+ return 1
+ print(f"VALID {package.package_id}")
+ print(f"as_of {package.payload['as_of']}")
+ print(f"admission {package.payload['admission']['decision']}")
+ print("authenticity not established; verify provenance through a trusted registry")
+ print("investment not approved; expert judgement remains required")
+ return 0
+
+
+def cmd_adapter_check(args) -> int:
+ from .data.conformance import check_adapter
+
+ qf = Quantifact(args.workspace, _user(args.user))
+ report = check_adapter(
+ qf.adapter,
+ early_as_of=args.early_as_of,
+ late_as_of=args.late_as_of,
+ sample_size=args.sample_size,
+ )
+ for check in report.checks:
+ print(f"{'PASS' if check.passed else 'FAIL'} {check.id:<34} {check.evidence}")
+ if args.json:
+ Path(args.json).write_text(json.dumps(report.to_dict(), indent=2))
+ print(f"wrote {args.json}")
+ return 0 if report.passed else 1
+
+
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(
prog="qf",
@@ -338,7 +389,18 @@ def main(argv: list[str] | None = None) -> int:
a.add_argument("--answers", help="JSON dict of clarification answers")
a.add_argument("--out", help="report path")
a.add_argument("--receipt", help="write a JSON receipt of this run")
+ a.add_argument(
+ "--evidence",
+ help="write the versioned research evidence package (default: beside report)",
+ )
a.add_argument("--no-cache", action="store_true")
+ a.add_argument(
+ "--execution",
+ choices=["in_process", "process"],
+ default="in_process",
+ help="process contains crashes/timeouts; it is not a no-network sandbox",
+ )
+ a.add_argument("--task-timeout", type=float, default=30.0)
a.add_argument("--backend", default="reference", choices=["reference", "llm"])
a.add_argument("--fix", action="store_true", help="enable the debugger agent")
a.add_argument("--semantic", action="store_true", help="enable L3 review")
@@ -393,6 +455,7 @@ def main(argv: list[str] | None = None) -> int:
e = sub.add_parser("evals")
e.set_defaults(fn=cmd_evals)
e.add_argument("--dir", help="benchmark directory (default: the workspace's)")
+ e.add_argument("--json", help="write the sliced machine-readable evaluation")
b = sub.add_parser("bench")
b.set_defaults(fn=cmd_bench)
@@ -409,8 +472,35 @@ def main(argv: list[str] | None = None) -> int:
help="exit non-zero until PAT-level evidence is reached",
)
+ verify = sub.add_parser("verify")
+ verify.set_defaults(fn=cmd_verify)
+ verify.add_argument("package", help="research evidence package JSON")
+
+ adapter = sub.add_parser("adapter-check")
+ adapter.set_defaults(fn=cmd_adapter_check)
+ adapter.add_argument("--early-as-of", default="2022-03-01")
+ adapter.add_argument("--late-as-of", default="2026-08-01")
+ adapter.add_argument("--sample-size", type=int, default=8)
+ adapter.add_argument("--json")
+
args = ap.parse_args(argv)
- return args.fn(args)
+ try:
+ return args.fn(args)
+ except Exception as exc:
+ # Domain refusals are expected product outcomes. Keep programming bugs
+ # noisy, but do not present an unsupported question or failed contract
+ # as an internal crash.
+ from .contracts.point_in_time import LookAheadError
+ from .contracts.verdict import TaskUnfixable
+ from .plan.model import PlanError
+ from .planner import UnsupportedQuestionError
+
+ if isinstance(
+ exc, (UnsupportedQuestionError, LookAheadError, PlanError, TaskUnfixable)
+ ):
+ print(f"refused: {exc}", file=sys.stderr)
+ return 2
+ raise
if __name__ == "__main__":
diff --git a/src/quantifact/contracts/methods.py b/src/quantifact/contracts/methods.py
new file mode 100644
index 0000000..c2e683c
--- /dev/null
+++ b/src/quantifact/contracts/methods.py
@@ -0,0 +1,141 @@
+"""Research-family contracts.
+
+Generic schemas can prove that a dataframe matches its declaration. They
+cannot prove that an event study declared its window before seeing results or
+that a historical analogy avoided causal overclaiming. Method contracts close
+that gap one research family at a time and remain ordinary Python gates: a
+model may propose a methodology, but it cannot waive its requirements.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import pandas as pd
+
+from ..plan.model import AnalysisPlan
+from .verdict import Verdict
+
+DesignCheck = Callable[[AnalysisPlan], list[str]]
+EvidenceCheck = Callable[[AnalysisPlan, dict[str, pd.DataFrame]], list[str]]
+
+
+def _event_study_design(plan: AnalysisPlan) -> list[str]:
+ problems: list[str] = []
+ event_tasks = [
+ t for t in plan.tasks if (t.op or {}).get("kind") == "event_window_return"
+ ]
+ if not event_tasks:
+ return ["event_study declares no event_window_return task"]
+ for task in event_tasks:
+ window = (task.op or {}).get("window_days")
+ if not isinstance(window, int) or window <= 0:
+ problems.append(
+ f"task '{task.name}' has no positive pre-declared event window"
+ )
+ exact = [
+ x
+ for x in task.invariants
+ if x.get("kind") == "row_count" and x.get("min") == x.get("max")
+ ]
+ if not exact:
+ problems.append(f"task '{task.name}' has no exact expected row-set contract")
+ unique = [x for x in task.invariants if x.get("kind") == "unique"]
+ if not unique:
+ problems.append(
+ f"task '{task.name}' does not declare event/entity uniqueness"
+ )
+ design = plan.research_design
+ falsifiers = " ".join(
+ f for claim in (design.claims if design else []) for f in claim.falsifiers
+ ).lower()
+ if "window" not in falsifiers:
+ problems.append("event_study has no pre-declared window-sensitivity falsifier")
+ return problems
+
+
+def _event_study_evidence(
+ plan: AnalysisPlan, frames: dict[str, pd.DataFrame]
+) -> list[str]:
+ problems: list[str] = []
+ for task in plan.tasks:
+ if (task.op or {}).get("kind") != "event_window_return":
+ continue
+ frame = frames.get(task.name)
+ if frame is None or frame.empty:
+ problems.append(f"event result '{task.name}' did not materialise")
+ continue
+ if {"episode", "market_id"} <= set(frame.columns):
+ counts = frame.groupby("episode")["market_id"].nunique()
+ if len(counts) < 2:
+ problems.append("event study materialised fewer than two episodes")
+ return problems
+
+
+def _historical_analogy_design(plan: AnalysisPlan) -> list[str]:
+ design = plan.research_design
+ if design is None:
+ return ["historical_analogy has no research design"]
+ problems: list[str] = []
+ if any(c.kind == "causal" for c in design.claims):
+ problems.append(
+ "historical analogy may not present an analogy as causal evidence"
+ )
+ limits = " ".join(design.limitations).lower()
+ if not any(word in limits for word in ("observational", "causal", "forecast")):
+ problems.append("historical analogy does not disclose its inference limit")
+ if not design.alternatives:
+ problems.append("historical analogy declares no rival explanation")
+ return problems
+
+
+def _historical_analogy_evidence(
+ plan: AnalysisPlan, frames: dict[str, pd.DataFrame]
+) -> list[str]:
+ episode_frames = [df for df in frames.values() if "episode" in df.columns]
+ if not episode_frames:
+ return ["historical analogy materialised no episode-labelled evidence"]
+ episodes = {str(x) for df in episode_frames for x in df["episode"].dropna().unique()}
+ return (
+ []
+ if len(episodes) >= 2
+ else ["historical analogy contains fewer than two episodes"]
+ )
+
+
+METHODS: dict[str, tuple[DesignCheck, EvidenceCheck]] = {
+ "event_study": (_event_study_design, _event_study_evidence),
+ "historical_analogy": (_historical_analogy_design, _historical_analogy_evidence),
+}
+
+
+def validate_method_design(plan: AnalysisPlan) -> list[str]:
+ design = plan.research_design
+ if design is None:
+ return []
+ problems: list[str] = []
+ for method in design.methodologies:
+ registered = METHODS.get(method)
+ if registered is None:
+ problems.append(f"unknown research methodology '{method}'")
+ else:
+ problems.extend(registered[0](plan))
+ return problems
+
+
+def validate_method_evidence(
+ plan: AnalysisPlan, frames: dict[str, pd.DataFrame]
+) -> list[Verdict]:
+ design = plan.research_design
+ if design is None:
+ return []
+ verdicts: list[Verdict] = []
+ for method in design.methodologies:
+ registered = METHODS.get(method)
+ problems = (
+ [f"unknown research methodology '{method}'"]
+ if registered is None
+ else registered[1](plan, frames)
+ )
+ verdicts.append(Verdict(method, "M-method", not problems, problems))
+ return verdicts
diff --git a/src/quantifact/data/adapters/demo_synthetic.py b/src/quantifact/data/adapters/demo_synthetic.py
index 7362da8..1bd0dad 100644
--- a/src/quantifact/data/adapters/demo_synthetic.py
+++ b/src/quantifact/data/adapters/demo_synthetic.py
@@ -388,11 +388,29 @@ class DemoSyntheticAdapter:
def __init__(self, root: str | Path, seed: int = 42):
root = Path(root)
- self.store = (
- SeriesStore(root)
- if (root / "catalog.json").exists()
- else build_store(root, seed)
- )
+ root.mkdir(parents=True, exist_ok=True)
+ # CLI commands may start concurrently against the same workspace. A
+ # build lock prevents one command from reading a partially populated
+ # catalog while another is still writing the demo Parquet files.
+ lock_path = root / ".build.lock"
+ with lock_path.open("a+") as lock:
+ try:
+ import fcntl
+
+ fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
+ except ImportError: # pragma: no cover - Windows is best effort
+ pass
+ existing = SeriesStore(root) if (root / "catalog.json").exists() else None
+ complete = existing is not None and all(
+ existing._path(series_id).exists() for series_id in existing.ids
+ )
+ self.store = existing if complete else build_store(root, seed)
+ try:
+ import fcntl
+
+ fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
+ except ImportError: # pragma: no cover
+ pass
self._universe = universe()[
["market_id", "asset_class", "listed_from", "delisted_on"]
]
diff --git a/src/quantifact/data/conformance.py b/src/quantifact/data/conformance.py
new file mode 100644
index 0000000..349d558
--- /dev/null
+++ b/src/quantifact/data/conformance.py
@@ -0,0 +1,154 @@
+"""Executable conformance contract for third-party point-in-time adapters."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import date
+from typing import Any
+
+import pandas as pd
+
+from ..plan.model import parse_date
+
+
+@dataclass
+class ConformanceCheck:
+ id: str
+ passed: bool
+ evidence: str
+
+
+@dataclass
+class AdapterConformanceReport:
+ adapter: str
+ checks: list[ConformanceCheck]
+
+ @property
+ def passed(self) -> bool:
+ return bool(self.checks) and all(check.passed for check in self.checks)
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "schema_version": 1,
+ "adapter": self.adapter,
+ "passed": self.passed,
+ "checks": [asdict(check) for check in self.checks],
+ }
+
+
+def check_adapter(
+ adapter: Any,
+ *,
+ early_as_of: str | date,
+ late_as_of: str | date,
+ sample_size: int = 8,
+) -> AdapterConformanceReport:
+ """Probe temporal claims every production adapter must make reproducibly.
+
+ Passing establishes protocol behaviour on sampled catalog entries. It does
+ not establish source correctness, licence compliance or full-catalog quality.
+ """
+ early = parse_date(early_as_of)
+ late = parse_date(late_as_of)
+ if early >= late:
+ raise ValueError("early_as_of must be before late_as_of")
+ checks: list[ConformanceCheck] = []
+
+ catalog = adapter.catalog()
+ ids = [meta.series_id for meta in catalog]
+ checks.append(ConformanceCheck("catalog_nonempty", bool(ids), f"{len(ids)} series"))
+ unique = len(ids) == len(set(ids))
+ checks.append(
+ ConformanceCheck(
+ "catalog_ids_unique",
+ unique,
+ "all identifiers unique" if unique else "duplicate series identifiers",
+ )
+ )
+ required = ("frequency", "unit", "source", "license_tag")
+ complete = all(
+ all(getattr(meta, field, None) for field in required) for meta in catalog
+ )
+ checks.append(
+ ConformanceCheck(
+ "catalog_semantics_complete",
+ complete,
+ f"required fields: {', '.join(required)}",
+ )
+ )
+
+ sampled = ids[:sample_size]
+ no_future = True
+ monotone = True
+ deterministic = True
+ read_errors: list[str] = []
+ for sid in sampled:
+ try:
+ early_values = adapter.read_series(sid, as_of=early)
+ late_values = adapter.read_series(sid, as_of=late)
+ except Exception as exc:
+ read_errors.append(f"{sid}: {type(exc).__name__}: {exc}")
+ no_future = monotone = deterministic = False
+ continue
+ for values, cut in ((early_values, early), (late_values, late)):
+ if not isinstance(values, pd.Series):
+ no_future = False
+ continue
+ if len(values) and pd.Timestamp(values.index.max()) > pd.Timestamp(cut):
+ no_future = False
+ if not set(early_values.index).issubset(set(late_values.index)):
+ monotone = False
+ a = adapter.fingerprint([sid], as_of=early)
+ b = adapter.fingerprint([sid], as_of=early)
+ if not a or a != b:
+ deterministic = False
+ checks += [
+ ConformanceCheck(
+ "sample_series_readable",
+ not read_errors,
+ (
+ f"sampled {len(sampled)} series"
+ if not read_errors
+ else "; ".join(read_errors[:3])
+ ),
+ ),
+ ConformanceCheck(
+ "no_future_observation_dates",
+ no_future,
+ f"sampled {len(sampled)} series at two vintages",
+ ),
+ ConformanceCheck(
+ "vintage_visibility_monotone",
+ monotone,
+ "early observation keys are a subset of later keys",
+ ),
+ ConformanceCheck(
+ "fingerprint_deterministic",
+ deterministic,
+ f"repeated fingerprints agree for {len(sampled)} series",
+ ),
+ ]
+
+ table_names = adapter.tables()
+ tables_ok = True
+ for name in table_names:
+ frame = adapter.read_table(name, as_of=early)
+ tables_ok = tables_ok and isinstance(frame, pd.DataFrame)
+ date_columns = [
+ column
+ for column in frame.columns
+ if pd.api.types.is_datetime64_any_dtype(frame[column])
+ ]
+ for column in date_columns:
+ if len(frame) and pd.Timestamp(frame[column].max()) > pd.Timestamp(early):
+ tables_ok = False
+ checks.append(
+ ConformanceCheck(
+ "reference_tables_point_in_time",
+ tables_ok,
+ f"checked {len(table_names)} reference tables",
+ )
+ )
+ return AdapterConformanceReport(
+ getattr(adapter, "name", type(adapter).__name__), checks
+ )
diff --git a/src/quantifact/data/registry.py b/src/quantifact/data/registry.py
index a2b8eaa..a423052 100644
--- a/src/quantifact/data/registry.py
+++ b/src/quantifact/data/registry.py
@@ -125,6 +125,7 @@ def __init__(self, root: str | Path):
self.data_dir.mkdir(parents=True, exist_ok=True)
self.index_path = self.root / "catalog.json"
self._meta: dict[str, SeriesMeta] = {}
+ self._visible_fingerprints: dict[tuple[str, str, str], bytes] = {}
if self.index_path.exists():
raw = json.loads(self.index_path.read_text())
self._meta = {k: SeriesMeta.from_dict(v) for k, v in raw.items()}
@@ -171,6 +172,13 @@ def write(
meta.content_hash = _content_hash(df)
df.to_parquet(self._path(meta.series_id))
self._meta[meta.series_id] = meta
+ # A write changes the physical series and may change any visible
+ # vintage at or after its publication dates.
+ self._visible_fingerprints = {
+ key: value
+ for key, value in self._visible_fingerprints.items()
+ if key[0] != meta.series_id
+ }
self.flush()
return meta
@@ -206,8 +214,21 @@ def fingerprint(self, series_ids: Iterable[str], *, as_of: str | date) -> str:
cache entry, because they are answering different questions.
"""
h = hashlib.sha256()
- h.update(str(parse_date(as_of)).encode())
+ cut = pd.Timestamp(parse_date(as_of))
+ h.update(str(cut.date()).encode())
for sid in sorted(series_ids):
h.update(sid.encode())
- h.update(self._meta[sid].content_hash.encode())
+ # Hash what this run was actually allowed to observe, not the full
+ # physical file. A later revision or publication must not alter
+ # the identity of an earlier vintage.
+ cache_id = (sid, str(cut.date()), self._meta[sid].content_hash)
+ visible_hash = self._visible_fingerprints.get(cache_id)
+ if visible_hash is None:
+ visible = self.read_frame(sid)
+ visible = visible[visible["pub_date"] <= cut]
+ visible_hash = hashlib.sha256(
+ pd.util.hash_pandas_object(visible, index=True).to_numpy().tobytes()
+ ).digest()
+ self._visible_fingerprints[cache_id] = visible_hash
+ h.update(visible_hash)
return h.hexdigest()[:16]
diff --git a/src/quantifact/evidence.py b/src/quantifact/evidence.py
new file mode 100644
index 0000000..2896d88
--- /dev/null
+++ b/src/quantifact/evidence.py
@@ -0,0 +1,197 @@
+"""The durable product of a Quantifact run.
+
+Reports are views and conversations are transient. The evidence package is the
+versioned, machine-verifiable research object: question, design, knowledge date,
+source vintages and licences, code identity, materialised outputs, verdicts,
+claim lineage and an explicit admission decision. Admission means only that
+the declared evidence crossed the configured gates; it is never investment
+approval or a claim that the inference is true.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+from .harness.cache import RUNTIME_ID, frame_fingerprint
+
+SCHEMA_VERSION = "quantifact.evidence/1"
+
+
+def _canonical(value: Any) -> str:
+ return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _digest(value: Any) -> str:
+ return hashlib.sha256(_canonical(value).encode()).hexdigest()
+
+
+@dataclass
+class ResearchEvidencePackage:
+ payload: dict[str, Any]
+
+ @property
+ def package_id(self) -> str:
+ return self.payload["integrity"]["sha256"]
+
+ @property
+ def admitted(self) -> bool:
+ return self.payload["admission"]["evidence_admitted"]
+
+ def verify(self) -> list[str]:
+ problems: list[str] = []
+ if self.payload.get("schema_version") != SCHEMA_VERSION:
+ problems.append(
+ f"unsupported schema_version {self.payload.get('schema_version')!r}"
+ )
+ integrity = self.payload.get("integrity", {})
+ body = {k: v for k, v in self.payload.items() if k != "integrity"}
+ expected = _digest(body)
+ if integrity.get("sha256") != expected:
+ problems.append("package integrity hash does not match its contents")
+ if not self.payload.get("as_of"):
+ problems.append("package has no knowledge date")
+ if not self.payload.get("claims"):
+ problems.append("package carries no claim lineage")
+ plan = self.payload.get("plan") or {}
+ plan_names = {t.get("name") for t in plan.get("tasks", [])}
+ task_names = set(self.payload.get("tasks", {}))
+ if plan_names != task_names:
+ problems.append("package task manifest does not match its plan")
+ codes = self.payload.get("code", {})
+ if set(codes) != task_names:
+ problems.append("package code manifest does not match its tasks")
+ for name, source in codes.items():
+ actual = hashlib.sha256(source.encode()).hexdigest()
+ expected_code = self.payload["tasks"].get(name, {}).get("code_sha256")
+ if actual != expected_code:
+ problems.append(f"task '{name}' code hash does not match embedded source")
+ return problems
+
+ def to_dict(self) -> dict[str, Any]:
+ return self.payload
+
+ def save(self, path: str | Path) -> Path:
+ target = Path(path)
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_text(json.dumps(self.payload, indent=2, ensure_ascii=False))
+ return target
+
+ @staticmethod
+ def load(path: str | Path) -> ResearchEvidencePackage:
+ return ResearchEvidencePackage(json.loads(Path(path).read_text()))
+
+
+def _task_sources(plan, name: str, memo: dict[str, list[str]]) -> list[str]:
+ if name in memo:
+ return memo[name]
+ task = plan[name]
+ sources = set(task.series_inputs)
+ for dep in task.depends_on:
+ sources.update(_task_sources(plan, dep, memo))
+ memo[name] = sorted(sources)
+ return memo[name]
+
+
+def build_evidence_package(
+ *,
+ plan,
+ codes: dict[str, str],
+ result,
+ findings,
+ verdicts,
+ timings: dict[str, float],
+ planning_trace: dict[str, Any],
+ repair_trace: list[dict[str, Any]],
+ adapter,
+ backend: str,
+ user: str,
+ report_path: Path | None,
+ execution_mode: str = "in_process",
+) -> ResearchEvidencePackage:
+ metadata = {m.series_id: m for m in adapter.catalog()}
+ source_ids = plan.series_inputs()
+ sources = []
+ for sid in source_ids:
+ meta = metadata[sid]
+ sources.append(
+ {
+ "series_id": sid,
+ "source": meta.source,
+ "license": meta.license_tag,
+ "frequency": meta.frequency,
+ "unit": meta.unit,
+ "first_observation": meta.first_obs,
+ "last_observation": meta.last_obs,
+ "last_publication": meta.last_pub,
+ "visible_fingerprint": adapter.fingerprint([sid], as_of=plan.as_of),
+ }
+ )
+
+ memo: dict[str, list[str]] = {}
+ tasks: dict[str, Any] = {}
+ for task in plan.tasks:
+ frame = result.frames[task.name]
+ tasks[task.name] = {
+ "type": task.type,
+ "depends_on": task.depends_on,
+ "source_series": _task_sources(plan, task.name, memo),
+ "code_sha256": hashlib.sha256(codes[task.name].encode()).hexdigest(),
+ "value_fingerprint": frame_fingerprint(frame),
+ "rows": len(frame),
+ "columns": list(frame.columns),
+ "cache_key": result.trace(task.name).cache_key,
+ }
+
+ design = plan.research_design
+ claims = []
+ if design:
+ for claim in design.claims:
+ claims.append(
+ {
+ **asdict(claim),
+ "evidence": {name: tasks[name] for name in claim.evidence_tasks},
+ }
+ )
+
+ admission = {
+ "evidence_admitted": True,
+ "decision": "admitted_for_expert_review",
+ "meaning": (
+ "All mandatory system gates completed. This is not investment approval, "
+ "proof that a claim is true, or authorisation to trade."
+ ),
+ "investment_approved": False,
+ "blocking_findings": [asdict(f) for f in findings if f.severity == "blocking"],
+ }
+ body = {
+ "schema_version": SCHEMA_VERSION,
+ "question": plan.question,
+ "as_of": plan.as_of,
+ "identity": {
+ "user": user,
+ "backend": backend,
+ "runtime": RUNTIME_ID,
+ "execution_mode": execution_mode,
+ },
+ "admission": admission,
+ "research_design": asdict(design) if design else None,
+ "plan": plan.to_dict(),
+ "code": dict(codes),
+ "resolved_assumptions": plan.resolved_assumptions,
+ "sources": sources,
+ "tasks": tasks,
+ "claims": claims,
+ "verdicts": [asdict(v) for v in verdicts],
+ "findings": [asdict(f) for f in findings],
+ "planning_trace": planning_trace,
+ "repair_trace": repair_trace,
+ "timings": dict(timings),
+ "report": str(report_path) if report_path else None,
+ }
+ return ResearchEvidencePackage(
+ {**body, "integrity": {"algorithm": "sha256", "sha256": _digest(body)}}
+ )
diff --git a/src/quantifact/harness/execute.py b/src/quantifact/harness/execute.py
index 14f6763..8ed9dea 100644
--- a/src/quantifact/harness/execute.py
+++ b/src/quantifact/harness/execute.py
@@ -14,6 +14,7 @@
from __future__ import annotations
import builtins
+import multiprocessing as mp
import time
from collections.abc import Callable
from dataclasses import dataclass
@@ -65,6 +66,68 @@
}
+def _namespace(adapter: Any, as_of: str) -> dict[str, Any]:
+ """The generated-code namespace, shared by in-process and worker execution."""
+
+ def load_series(series_id: str) -> pd.Series:
+ return adapter.read_series(series_id, as_of=as_of)
+
+ def load_table(name: str) -> pd.DataFrame:
+ return adapter.read_table(name, as_of=as_of)
+
+ return {
+ "__builtins__": SAFE_BUILTINS,
+ "pd": pd,
+ "np": np,
+ "load_series": load_series,
+ "load_table": load_table,
+ }
+
+
+def _execute_source(
+ adapter: Any, task_name: str, source: str, args: list[pd.DataFrame], as_of: str
+) -> pd.DataFrame:
+ ns = _namespace(adapter, as_of)
+ exec(compile(source, f"", "exec"), ns)
+ fn = ns.get(task_name)
+ if not callable(fn):
+ raise RuntimeError(f"generated code defines no function '{task_name}'")
+ result = fn(*args)
+ if not isinstance(result, pd.DataFrame):
+ raise TypeError(
+ f"{task_name} returned {type(result).__name__}, expected DataFrame"
+ )
+ return result.reset_index(drop=True)
+
+
+def _process_worker(
+ connection,
+ adapter: Any,
+ task_name: str,
+ source: str,
+ args: list[pd.DataFrame],
+ as_of: str,
+ cpu_seconds: int,
+ memory_mb: int,
+) -> None:
+ """Child entry point. Resource limits are best-effort and platform-specific."""
+ try:
+ try:
+ import resource
+
+ resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds + 1))
+ if hasattr(resource, "RLIMIT_AS"):
+ limit = memory_mb * 1024 * 1024
+ resource.setrlimit(resource.RLIMIT_AS, (limit, limit))
+ except (ImportError, OSError, ValueError):
+ pass
+ connection.send((True, _execute_source(adapter, task_name, source, args, as_of)))
+ except BaseException as exc: # child must return a bounded error, then die
+ connection.send((False, f"{type(exc).__name__}: {exc}"))
+ finally:
+ connection.close()
+
+
@dataclass
class TaskTrace:
task: str
@@ -112,21 +175,7 @@ def __init__(self, adapter: Any, cache: ValueCache):
def _namespace(self, as_of: str) -> dict[str, Any]:
"""Loaders bound to the knowledge date. Generated code cannot rebind
them: static analysis rejects any keyword argument to a loader."""
- adapter = self.adapter
-
- def load_series(series_id: str) -> pd.Series:
- return adapter.read_series(series_id, as_of=as_of)
-
- def load_table(name: str) -> pd.DataFrame:
- return adapter.read_table(name, as_of=as_of)
-
- return {
- "__builtins__": SAFE_BUILTINS,
- "pd": pd,
- "np": np,
- "load_series": load_series,
- "load_table": load_table,
- }
+ return _namespace(self.adapter, as_of)
def _compile(
self, task_name: str, source: str, as_of: str
@@ -153,13 +202,14 @@ def run_one(
) -> pd.DataFrame:
"""Execute a single task against already-materialised upstream frames.
Used by the repair loop and by graders; no caching, no layering."""
- fn = self._compile(task.name, source, as_of)
- df = fn(*[frames[d] for d in task.depends_on])
- if not isinstance(df, pd.DataFrame):
- raise TypeError(
- f"{task.name} returned {type(df).__name__}, expected DataFrame"
- )
- return df.reset_index(drop=True)
+ return self._materialise(
+ task, source, [frames[d] for d in task.depends_on], as_of
+ )
+
+ def _materialise(
+ self, task: Task, source: str, args: list[pd.DataFrame], as_of: str
+ ) -> pd.DataFrame:
+ return _execute_source(self.adapter, task.name, source, args, as_of)
# ----------------------------------------------------------------- run
def run(
@@ -203,9 +253,13 @@ def run(
list(cached.columns),
)
else:
- fn = self._compile(name, codes[name], as_of)
try:
- df = fn(*[frames[d] for d in task.depends_on])
+ df = self._materialise(
+ task,
+ codes[name],
+ [frames[d] for d in task.depends_on],
+ as_of,
+ )
except Exception as e:
tr = TaskTrace(
name,
@@ -242,3 +296,65 @@ def run(
on_task(tr)
return RunResult(frames=frames, traces=traces, layers=layers, as_of=as_of)
+
+
+class ProcessExecutionHarness(ExecutionHarness):
+ """Run each generated task in a disposable, resource-bounded process.
+
+ This contains interpreter crashes, runaway CPU and wall-time overruns. It is
+ intentionally not called a sandbox: the worker still shares the host kernel
+ and network namespace. Production deployment must put this worker behind a
+ no-network container or VM boundary.
+ """
+
+ def __init__(
+ self,
+ adapter: Any,
+ cache: ValueCache,
+ *,
+ timeout_seconds: float = 10.0,
+ cpu_seconds: int = 5,
+ memory_mb: int = 1024,
+ ):
+ super().__init__(adapter, cache)
+ self.timeout_seconds = timeout_seconds
+ self.cpu_seconds = cpu_seconds
+ self.memory_mb = memory_mb
+
+ def _materialise(
+ self, task: Task, source: str, args: list[pd.DataFrame], as_of: str
+ ) -> pd.DataFrame:
+ context = mp.get_context("spawn")
+ parent, child = context.Pipe(duplex=False)
+ process = context.Process(
+ target=_process_worker,
+ args=(
+ child,
+ self.adapter,
+ task.name,
+ source,
+ args,
+ as_of,
+ self.cpu_seconds,
+ self.memory_mb,
+ ),
+ )
+ process.start()
+ child.close()
+ process.join(self.timeout_seconds)
+ if process.is_alive():
+ process.terminate()
+ process.join(1)
+ parent.close()
+ raise TimeoutError(
+ f"task '{task.name}' exceeded {self.timeout_seconds:.2f}s wall limit"
+ )
+ if not parent.poll():
+ code = process.exitcode
+ parent.close()
+ raise RuntimeError(f"task worker exited without a result (exitcode={code})")
+ ok, value = parent.recv()
+ parent.close()
+ if not ok:
+ raise RuntimeError(value)
+ return value
diff --git a/src/quantifact/learn/benchmarks.py b/src/quantifact/learn/benchmarks.py
index 1b4e5ae..0465955 100644
--- a/src/quantifact/learn/benchmarks.py
+++ b/src/quantifact/learn/benchmarks.py
@@ -25,6 +25,10 @@ class Benchmark:
origin: str = "teach"
human_audited: bool = False
note: str = ""
+ family: str = "unspecified"
+ risk_tags: list[str] = field(default_factory=list)
+ expected_outcome: str = "plan" # plan | refuse
+ severity: str = "major" # critical | major | minor
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@@ -41,16 +45,82 @@ class BenchmarkResult:
failures: list[str] = field(default_factory=list)
+@dataclass
+class BenchmarkReport:
+ results: list[BenchmarkResult]
+
+ @property
+ def passed(self) -> int:
+ return sum(r.passed for r in self.results)
+
+ @property
+ def total(self) -> int:
+ return len(self.results)
+
+ @property
+ def silent_critical_failures(self) -> int:
+ return sum(
+ not r.passed
+ and r.benchmark.severity == "critical"
+ and r.benchmark.expected_outcome != "refuse"
+ for r in self.results
+ )
+
+ def slices(self, field: str) -> dict[str, dict[str, int]]:
+ grouped: dict[str, list[BenchmarkResult]] = {}
+ for result in self.results:
+ values = getattr(result.benchmark, field)
+ values = values if isinstance(values, list) else [values]
+ for value in values or ["unspecified"]:
+ grouped.setdefault(value, []).append(result)
+ return {
+ key: {"passed": sum(r.passed for r in rows), "total": len(rows)}
+ for key, rows in sorted(grouped.items())
+ }
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "passed": self.passed,
+ "total": self.total,
+ "silent_critical_failures": self.silent_critical_failures,
+ "by_family": self.slices("family"),
+ "by_risk": self.slices("risk_tags"),
+ "results": [
+ {
+ "id": r.benchmark.id,
+ "family": r.benchmark.family,
+ "expected_outcome": r.benchmark.expected_outcome,
+ "severity": r.benchmark.severity,
+ "risk_tags": r.benchmark.risk_tags,
+ "passed": r.passed,
+ "failures": r.failures,
+ }
+ for r in self.results
+ ],
+ }
+
+
class BenchmarkSuite:
def __init__(self, root: str | Path):
self.root = Path(root)
self.root.mkdir(parents=True, exist_ok=True)
def all(self) -> list[Benchmark]:
- return [
- Benchmark.from_dict(json.loads(p.read_text()))
- for p in sorted(self.root.glob("*.json"))
- ]
+ benchmarks = []
+ for path in sorted(self.root.rglob("*.json")):
+ payload = json.loads(path.read_text())
+ # Benchmark directories may also contain performance runs, model
+ # trials, and operating-evidence templates. Only documents with
+ # the benchmark identity are cases; once identified, malformed
+ # fields still fail loudly in ``from_dict``.
+ if not isinstance(payload, dict) or not {
+ "id",
+ "prompt",
+ "assertions",
+ }.issubset(payload):
+ continue
+ benchmarks.append(Benchmark.from_dict(payload))
+ return benchmarks
def add(self, bench: Benchmark) -> Path:
p = self.root / f"{bench.id}.json"
@@ -76,10 +146,19 @@ def run(
try:
plan = planner.plan(bench.prompt, bench.answers)
except Exception as e:
+ if bench.expected_outcome == "refuse":
+ return BenchmarkResult(bench, True)
return BenchmarkResult(
bench, False, [f"planning raised {type(e).__name__}: {e}"]
)
+ if bench.expected_outcome == "refuse":
+ return BenchmarkResult(
+ bench,
+ False,
+ ["planner produced a plan for a question that must be refused"],
+ )
+
for a in bench.assertions:
t = a["type"]
if t == "plan_has_task":
@@ -142,3 +221,8 @@ def run_all(
self, adapter, lessons: list[Lesson], entitlements: tuple[str, ...] = ()
) -> list[BenchmarkResult]:
return [self.run(b, adapter, lessons, entitlements) for b in self.all()]
+
+ def report(
+ self, adapter, lessons: list[Lesson], entitlements: tuple[str, ...] = ()
+ ) -> BenchmarkReport:
+ return BenchmarkReport(self.run_all(adapter, lessons, entitlements))
diff --git a/src/quantifact/plan/compile.py b/src/quantifact/plan/compile.py
index 736a690..ca965e1 100644
--- a/src/quantifact/plan/compile.py
+++ b/src/quantifact/plan/compile.py
@@ -19,6 +19,7 @@
from __future__ import annotations
+from ..contracts.methods import validate_method_design
from ..contracts.reasoning import validate_research_design
from .layers import topo_layers
from .model import ALLOWED_DTYPES, ALLOWED_ROLES, AnalysisPlan, PlanError, parse_date
@@ -85,6 +86,7 @@ def validate(self, plan: AnalysisPlan) -> list[str]:
if plan.research_design is not None or self.require_research_design:
p.extend(validate_research_design(plan))
+ p.extend(validate_method_design(plan))
if not plan.tasks:
p.append("plan has no tasks")
diff --git a/src/quantifact/plan/model.py b/src/quantifact/plan/model.py
index b5b24a0..2eddc3b 100644
--- a/src/quantifact/plan/model.py
+++ b/src/quantifact/plan/model.py
@@ -89,6 +89,11 @@ class ResearchDesign:
limitations: list[str]
identification_strategy: str | None = None
out_of_sample_test: str | None = None
+ # Research families activate deterministic, domain-specific design and
+ # evidence contracts. This is deliberately separate from question_type:
+ # a comparative question may be an event study, a cross-sectional study,
+ # or something else with materially different failure modes.
+ methodologies: list[str] = field(default_factory=list)
@staticmethod
def from_dict(d: dict[str, Any]) -> ResearchDesign:
diff --git a/src/quantifact/planner.py b/src/quantifact/planner.py
index 0fdfb5c..571c2c4 100644
--- a/src/quantifact/planner.py
+++ b/src/quantifact/planner.py
@@ -82,6 +82,26 @@ class BindingTrace:
)
+class UnsupportedQuestionError(ValueError):
+ """The deterministic planner has no audited workflow for this question.
+
+ Refusal is a correctness outcome. Reusing the one exact rule workflow for
+ an unrelated question would produce a beautifully contracted answer to the
+ wrong problem, which is more dangerous than an explicit limitation.
+ """
+
+
+_OIL_TERMS = {
+ "oil",
+ "supply shock",
+ "energy shock",
+ "middle east",
+ "hormuz",
+ "macro conditions",
+}
+_EVENT_TERMS = {"respond", "response", "compare", "episode", "event", "shock"}
+
+
class RulePlanner:
def __init__(
self,
@@ -94,8 +114,26 @@ def __init__(
self.lessons = lessons or []
self.bindings: list[BindingTrace] = []
+ def supports(self, prompt: str) -> bool:
+ """Whether the exact offline workflow is admissible for ``prompt``."""
+ if not prompt.strip():
+ return True # internal call used only to obtain defaults
+ text = prompt.lower()
+ return any(term in text for term in _OIL_TERMS) and any(
+ term in text for term in _EVENT_TERMS
+ )
+
+ def require_supported(self, prompt: str) -> None:
+ if not self.supports(prompt):
+ raise UnsupportedQuestionError(
+ "the rule planner supports oil/energy event studies and historical "
+ "analogies only; use --planner llm for compiler-bounded planning or "
+ "add an audited workflow and benchmark for this research family"
+ )
+
# ------------------------------------------------------------- clarify
def clarify(self, prompt: str) -> list[Clarification]:
+ self.require_supported(prompt)
episodes = [
row["episode"]
for _, row in self.adapter.read_table(
@@ -249,6 +287,7 @@ def bind_universe(self, as_of: str) -> dict[str, str]:
# ---------------------------------------------------------------- plan
def plan(self, prompt: str, answers: dict[str, Any] | None = None) -> AnalysisPlan:
+ self.require_supported(prompt)
given = dict(answers or {})
a = {**self.defaults(), **given}
as_of: str = str(a["as_of"])
@@ -817,6 +856,7 @@ def plan(self, prompt: str, answers: dict[str, Any] | None = None) -> AnalysisPl
design = ResearchDesign(
question_type="comparative",
+ methodologies=["event_study", "historical_analogy"],
decision_context=(
"Assess whether the latest oil-supply episode is a useful historical "
"analogue; this analysis informs further investigation, not a trade."
diff --git a/src/quantifact/planner_llm.py b/src/quantifact/planner_llm.py
index 60ade2d..065cf32 100644
--- a/src/quantifact/planner_llm.py
+++ b/src/quantifact/planner_llm.py
@@ -45,6 +45,7 @@
"resolved_assumptions": [""],
"research_design": {
"question_type": "descriptive | comparative | causal | predictive",
+ "methodologies": ["event_study | historical_analogy"],
"decision_context": "",
"claims": [{
"id": "snake_case", "statement": "",
@@ -94,6 +95,7 @@
Hard rules, each one checked by the compiler before any code is written:
- pre-register bounded claims, falsifiers, rival explanations and evidence tasks
+- declare a registered methodology only when its deterministic method contracts apply
- comparative/causal/predictive work must test at least one rival explanation
- never label a claim causal without a causal question and identification strategy
- never label a claim predictive without a stated out-of-sample test
diff --git a/src/quantifact/quality.py b/src/quantifact/quality.py
index c68bea6..8a1392e 100644
--- a/src/quantifact/quality.py
+++ b/src/quantifact/quality.py
@@ -253,6 +253,8 @@ def _auto_probes(workspace: Path) -> dict[str, tuple[float, str, str]]:
"""Cheap executable evidence. A failed probe scores zero and stays legible."""
from .agent import Quantifact
from .data.adapters.base import DocumentSource
+ from .data.conformance import check_adapter
+ from .learn.benchmarks import Benchmark, BenchmarkSuite
from .learn.teach import KNOWN_EFFECTS
from .learn.workflows import WorkflowRepo
from .planner_llm import LLMPlanner
@@ -282,8 +284,8 @@ def _auto_probes(workspace: Path) -> dict[str, tuple[float, str, str]]:
and receipt["code_sha256"]
)
out["diagnosability"] = (
- 1 if trace_ok else 0.5,
- "versioned receipt covers plan/code identity, planning, execution and checks",
+ 1 if trace_ok and art.evidence and not art.evidence.verify() else 0.5,
+ "verifiable evidence package covers plan/code, sources, claims, execution and checks",
"complete the machine-readable run receipt" if not trace_ok else "",
)
# Two vintages must expose different slices, and future documents must stay hidden.
@@ -300,10 +302,14 @@ def _auto_probes(workspace: Path) -> dict[str, tuple[float, str, str]]:
"numeric and document vintages probed",
"add/fix bitemporal enforcement" if not pit else "",
)
+ adapter_report = check_adapter(
+ qf.adapter, early_as_of="2022-03-01", late_as_of="2026-08-01"
+ )
out["structured_data"] = (
- 0.7,
+ 0.75 if adapter_report.passed else 0.4,
f"{len(qf.adapter.catalog())} synthetic series; "
- "inspection and entitlements implemented",
+ f"{sum(c.passed for c in adapter_report.checks)}/{len(adapter_report.checks)} "
+ "adapter conformance checks passed",
"validate against a licensed production catalog",
)
docs = getattr(qf.adapter, "documents", None)
@@ -313,9 +319,22 @@ def _auto_probes(workspace: Path) -> dict[str, tuple[float, str, str]]:
"connect licensed corpora, web retrieval, chunking and retrieval evals",
)
workflows = WorkflowRepo().all()
+ refusal_suite = BenchmarkSuite(workspace / "audit-refusals")
+ refusal_suite.add(
+ Benchmark(
+ id="unsupported-family",
+ prompt="Build a DCF valuation for a semiconductor company",
+ assertions=[],
+ family="equity_fundamental",
+ risk_tags=["unsupported_family"],
+ expected_outcome="refuse",
+ severity="critical",
+ )
+ )
+ refusal_ok = refusal_suite.report(qf.adapter, []).passed == 1
out["dynamic_planning"] = (
- 0.5 if LLMPlanner else 0,
- "compiler-feedback planner implemented; stub-tested",
+ 0.6 if LLMPlanner and refusal_ok else 0,
+ "compiler-feedback planner stub-tested; unsupported rule workflows refuse",
"run real-model, multi-domain held-out planner evaluations",
)
out["feedback_gate"] = (
diff --git a/src/quantifact/report/render.py b/src/quantifact/report/render.py
index b852b96..b47ac10 100644
--- a/src/quantifact/report/render.py
+++ b/src/quantifact/report/render.py
@@ -435,6 +435,13 @@ def render_report(
parts.append(_kpi(k, str(v)))
parts.append("