163 lines
6.9 KiB
Python
163 lines
6.9 KiB
Python
|
"""Frontend research adapter for infospace-bench; one immutable source per run."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import shutil
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
from infospace_bench import create_infospace, load_infospace, register_artifact
|
||
|
|
from infospace_bench.workflow import FixtureAssistedGenerationAdapter, run_workflow
|
||
|
|
from infospace_bench.openrouter import OpenRouterAssistedGenerationAdapter
|
||
|
|
|
||
|
|
REPO = Path(__file__).resolve().parents[1]
|
||
|
|
PROFILE = REPO / "research/bench/profile"
|
||
|
|
WORKFLOW = "frontend-source-investigation"
|
||
|
|
|
||
|
|
|
||
|
|
def digest(path):
|
||
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def source_metadata(path):
|
||
|
|
text = Path(path).read_text()
|
||
|
|
parts = text.split("---", 2)
|
||
|
|
if len(parts) != 3 or parts[0].strip():
|
||
|
|
raise ValueError("Source needs YAML frontmatter and normalized Markdown text")
|
||
|
|
data = yaml.safe_load(parts[1])
|
||
|
|
required = ("title", "author", "locator", "version", "accessed", "evidence_kind")
|
||
|
|
if not isinstance(data, dict) or any(not data.get(k) for k in required):
|
||
|
|
raise ValueError(f"Required source metadata: {', '.join(required)}")
|
||
|
|
if data["evidence_kind"] not in ("primary", "secondary", "synthetic"):
|
||
|
|
raise ValueError("evidence_kind must be primary, secondary or synthetic")
|
||
|
|
if not parts[2].strip():
|
||
|
|
raise ValueError("Source body is empty")
|
||
|
|
return data
|
||
|
|
|
||
|
|
|
||
|
|
def initialize(source, workspace, slug):
|
||
|
|
source = Path(source).resolve()
|
||
|
|
metadata = source_metadata(source)
|
||
|
|
space = create_infospace(workspace, slug, name=metadata["title"], topic_domain="infotech")
|
||
|
|
shutil.copytree(PROFILE, space.root / "profile")
|
||
|
|
config_path = space.root / "infospace.yaml"
|
||
|
|
config = yaml.safe_load(config_path.read_text())
|
||
|
|
config["workflows"] = yaml.safe_load((PROFILE / "workflow.yaml").read_text())["workflows"]
|
||
|
|
config["viability"] = {} # Structural metrics are not a research acceptance gate.
|
||
|
|
config_path.write_text(yaml.safe_dump(config, sort_keys=False))
|
||
|
|
target = space.root / "artifacts/sources/source.md"
|
||
|
|
shutil.copyfile(source, target)
|
||
|
|
register_artifact(space.root, artifact_id="source/source.md", path=target,
|
||
|
|
kind="source", title=metadata["title"], provenance={
|
||
|
|
"sha256": digest(target), "bibliography": metadata,
|
||
|
|
"evidence_kind": metadata["evidence_kind"],
|
||
|
|
})
|
||
|
|
return space.root
|
||
|
|
|
||
|
|
|
||
|
|
def plan(root):
|
||
|
|
root = Path(root).resolve()
|
||
|
|
space = load_infospace(root)
|
||
|
|
sources = [a for a in space.artifacts if a.kind == "source"]
|
||
|
|
if len(sources) != 1:
|
||
|
|
raise ValueError("Exactly one source per workspace is required for stage isolation")
|
||
|
|
source = sources[0]
|
||
|
|
if digest(root / source.path) != source.provenance.get("sha256"):
|
||
|
|
raise ValueError("Source changed since intake; initialize a new workspace")
|
||
|
|
workflows = [w for w in space.config.workflows if w["id"] == WORKFLOW]
|
||
|
|
if len(workflows) != 1:
|
||
|
|
raise ValueError("Missing frontend research workflow")
|
||
|
|
stages = workflows[0]["stages"]
|
||
|
|
if any(s["input"] != "source" for s in stages):
|
||
|
|
raise ValueError("All stages must use the isolated source input")
|
||
|
|
return {"root": str(root), "workflow": WORKFLOW,
|
||
|
|
"source_sha256": source.provenance["sha256"],
|
||
|
|
"evidence_kind": source.provenance["evidence_kind"],
|
||
|
|
"stages": [s["id"] for s in stages],
|
||
|
|
"provider_calls": sum(s["kind"] == "assisted" for s in stages),
|
||
|
|
"cost_usd": "unknown; call limit is not a dollar budget",
|
||
|
|
"review_status": "unreviewed"}
|
||
|
|
|
||
|
|
|
||
|
|
class LimitedAdapter:
|
||
|
|
def __init__(self, adapter, limit):
|
||
|
|
self.adapter, self.limit, self.calls = adapter, limit, 0
|
||
|
|
|
||
|
|
def generate(self, request):
|
||
|
|
if self.calls >= self.limit:
|
||
|
|
raise ValueError("Provider call limit reached")
|
||
|
|
self.calls += 1
|
||
|
|
return self.adapter.generate(request)
|
||
|
|
|
||
|
|
|
||
|
|
def execute(root, adapter, max_calls=4):
|
||
|
|
root = Path(root).resolve()
|
||
|
|
manifest = plan(root)
|
||
|
|
if manifest["provider_calls"] > max_calls:
|
||
|
|
raise ValueError("Planned provider calls exceed --max-calls")
|
||
|
|
receipt = root / "reports/research-run.json"
|
||
|
|
if receipt.exists():
|
||
|
|
raise ValueError("Run already attempted; initialize a new workspace to preserve evidence")
|
||
|
|
# Record inputs before execution, including on failed or interrupted runs.
|
||
|
|
manifest["input_digests"] = {
|
||
|
|
str(p.relative_to(root)): digest(p)
|
||
|
|
for p in [root / "infospace.yaml", *sorted((root / "profile").rglob("*.md")),
|
||
|
|
root / "profile/workflow.yaml"]
|
||
|
|
}
|
||
|
|
manifest["status"] = "started"
|
||
|
|
receipt.write_text(json.dumps(manifest, indent=2) + "\n")
|
||
|
|
limited = LimitedAdapter(adapter, max_calls)
|
||
|
|
try:
|
||
|
|
result = run_workflow(root, WORKFLOW, assisted_adapter=limited)
|
||
|
|
manifest.update(status=result.status, run_record=result.run_record_path)
|
||
|
|
except Exception as exc:
|
||
|
|
manifest.update(status="failed", error_type=type(exc).__name__)
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
manifest["provider_calls_attempted"] = limited.calls
|
||
|
|
receipt.write_text(json.dumps(manifest, indent=2) + "\n")
|
||
|
|
return manifest
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
||
|
|
init = commands.add_parser("init")
|
||
|
|
init.add_argument("source", type=Path)
|
||
|
|
init.add_argument("--workspace", type=Path, default=REPO / "research/bench")
|
||
|
|
init.add_argument("--slug", required=True)
|
||
|
|
preview = commands.add_parser("plan")
|
||
|
|
preview.add_argument("root", type=Path)
|
||
|
|
run = commands.add_parser("run")
|
||
|
|
run.add_argument("root", type=Path)
|
||
|
|
providers = run.add_mutually_exclusive_group(required=True)
|
||
|
|
providers.add_argument("--fixture-responses", type=Path)
|
||
|
|
providers.add_argument("--provider", choices=["openrouter"])
|
||
|
|
run.add_argument("--model")
|
||
|
|
run.add_argument("--max-calls", type=int, default=4)
|
||
|
|
args = parser.parse_args()
|
||
|
|
try:
|
||
|
|
if args.command == "init":
|
||
|
|
result = {"root": str(initialize(args.source, args.workspace, args.slug))}
|
||
|
|
elif args.command == "plan":
|
||
|
|
result = plan(args.root)
|
||
|
|
else:
|
||
|
|
if args.fixture_responses:
|
||
|
|
if plan(args.root)["evidence_kind"] != "synthetic":
|
||
|
|
raise ValueError("Fixture responses may only run against synthetic sources")
|
||
|
|
adapter = FixtureAssistedGenerationAdapter.from_file(args.fixture_responses)
|
||
|
|
else:
|
||
|
|
if not args.model:
|
||
|
|
raise ValueError("Live generation requires --model")
|
||
|
|
adapter = OpenRouterAssistedGenerationAdapter(model=args.model, retry_limit=0)
|
||
|
|
result = execute(args.root, adapter, args.max_calls)
|
||
|
|
print(json.dumps(result, indent=2, default=str))
|
||
|
|
except (ValueError, OSError) as exc:
|
||
|
|
parser.exit(1, f"{exc}\n")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|