tegwick
cfa249ab13
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0cffe-206f-7323-89d3-b20ca4cf4bf9
304 lines
16 KiB
Python
304 lines
16 KiB
Python
"""Frontend research adapter for infospace-bench; one immutable source per run."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from urllib import error, request
|
|
from urllib.parse import urlsplit
|
|
from pathlib import Path
|
|
import shutil
|
|
|
|
import yaml
|
|
from infospace_bench import create_infospace, load_infospace, register_artifact
|
|
from infospace_bench.errors import InfospaceError
|
|
from infospace_bench.workflow import FixtureAssistedGenerationAdapter, run_workflow
|
|
from infospace_bench.workflow import AssistedGenerationResult
|
|
from llm_connect.models import RunConfig
|
|
from research_execution import ResearchExecution
|
|
|
|
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": (3 if (root / "profile/execution.json").exists() else
|
|
sum(s["kind"] == "assisted" for s in stages)),
|
|
"provider_calls_minimum": 2 if (root / "profile/execution.json").exists() else 4,
|
|
"execution_policy": "conditional-v1" if (root / "profile/execution.json").exists() else "legacy",
|
|
"cost_usd": "unknown; call limit is not a dollar budget",
|
|
"review_status": "unreviewed"}
|
|
|
|
|
|
def diagnostic_summary(data):
|
|
"""Allowlist numeric usage/settings; never persist raw debug data."""
|
|
debug = data.get("debug") or {}
|
|
payload = (debug.get("provider_request") or {}).get("payload") or {}
|
|
body = (debug.get("provider_response") or {}).get("body") or {}
|
|
usage = body.get("usage") or {}
|
|
settings = {key: payload[key] for key in ("temperature", "max_tokens", "top_p")
|
|
if type(payload.get(key)) in (int, float)}
|
|
settings["reasoning_effort_present"] = "reasoning_effort" in payload
|
|
settings["reasoning_present"] = "reasoning" in payload
|
|
counts = {key: usage[key] for key in ("prompt_tokens", "completion_tokens", "total_tokens", "cost")
|
|
if type(usage.get(key)) in (int, float)}
|
|
details = usage.get("completion_tokens_details") or {}
|
|
counts["completion_tokens_details"] = {key: details[key] for key in
|
|
("reasoning_tokens", "accepted_prediction_tokens", "rejected_prediction_tokens")
|
|
if type(details.get(key)) in (int, float)}
|
|
return {"available": bool(debug), "effective_settings": settings, "provider_usage": counts}
|
|
|
|
|
|
class LLMConnectAdapter:
|
|
"""Client of llm-connect's /execute contract; credentials stay at the service."""
|
|
|
|
def __init__(self, endpoint, model, max_tokens=4000, reasoning_effort=None, diagnostics=False):
|
|
parsed = urlsplit(endpoint)
|
|
if (parsed.scheme not in ("http", "https") or not parsed.hostname
|
|
or parsed.username or parsed.password or parsed.query or parsed.fragment):
|
|
raise ValueError("Use an HTTP(S) llm-connect URL without credentials or query")
|
|
if max_tokens <= 0:
|
|
raise ValueError("--max-tokens must be positive")
|
|
self.endpoint, self.model = endpoint.rstrip("/"), model
|
|
self.config = RunConfig(model_name=model, temperature=0.2, max_tokens=max_tokens)
|
|
if reasoning_effort is not None:
|
|
self.config.model_params["reasoning_effort"] = reasoning_effort
|
|
self.rejected_content = None
|
|
self.diagnostics = diagnostics
|
|
self.responses = []
|
|
|
|
def generate(self, generation):
|
|
payload = {"prompt": generation.prompt, "config": self.config.to_dict()}
|
|
req = request.Request(self.endpoint + ("/execute?debug=1" if self.diagnostics else "/execute"),
|
|
data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
try:
|
|
with request.urlopen(req, timeout=self.config.timeout_seconds + 15) as response:
|
|
data = json.load(response)
|
|
except error.HTTPError as exc:
|
|
raise InfospaceError("llm_connect_http", "llm-connect request failed",
|
|
{"error": f"HTTP Error {exc.code}"}) from None
|
|
except (error.URLError, TimeoutError):
|
|
raise InfospaceError("llm_connect_transport", "llm-connect unavailable", {}) from None
|
|
except ValueError:
|
|
raise InfospaceError("llm_connect_response", "Invalid llm-connect JSON", {}) from None
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Invalid llm-connect response object")
|
|
metadata = {key: data.get(key) for key in ("model", "usage", "finish_reason")}
|
|
extra = data.get("metadata") or {}
|
|
metadata.update({key: extra[key] for key in (
|
|
"provider", "response_id", "latency_seconds", "profile", "profile_model",
|
|
"profile_provider") if key in extra})
|
|
if self.diagnostics:
|
|
metadata["diagnostics"] = diagnostic_summary(data)
|
|
content = data.get("content")
|
|
metadata["content_characters"] = len(content) if isinstance(content, str) else None
|
|
metadata["content_sha256"] = (hashlib.sha256(content.encode()).hexdigest()
|
|
if isinstance(content, str) else None)
|
|
self.responses.append({"stage": generation.stage_id, **metadata})
|
|
if (not isinstance(data.get("content"), str) or not data["content"].strip()
|
|
or data.get("finish_reason") != "stop"):
|
|
self.rejected_content = content if isinstance(content, str) else None
|
|
raise ValueError("llm-connect returned empty or incomplete generation; see receipt")
|
|
return AssistedGenerationResult(markdown=data["content"], provider="llm-connect",
|
|
metadata=metadata)
|
|
|
|
|
|
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, reuse_from=None):
|
|
root = Path(root).resolve()
|
|
manifest = plan(root)
|
|
if max_calls < 0:
|
|
raise ValueError("--max-calls must be nonnegative")
|
|
if manifest["execution_policy"] == "legacy" and 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", *sorted((root / "profile").glob("*.json"))]
|
|
}
|
|
manifest["status"] = "started"
|
|
manifest["adapter"] = type(adapter).__name__
|
|
manifest["model"] = getattr(adapter, "model", "")
|
|
if isinstance(adapter, LLMConnectAdapter):
|
|
manifest["execution"] = {"endpoint": adapter.endpoint,
|
|
"config": adapter.config.to_dict(),
|
|
"diagnostics_requested": adapter.diagnostics,
|
|
"call_limit_scope": "service requests; server retries unknown"}
|
|
limited = LimitedAdapter(adapter, max_calls)
|
|
execution = None
|
|
if manifest["execution_policy"] == "conditional-v1":
|
|
policy = json.loads((root / "profile/execution.json").read_text())
|
|
if policy != {"policy": "conditional-v1"}:
|
|
raise ValueError("Unsupported execution policy")
|
|
execution = ResearchExecution(limited, {
|
|
"source": manifest["source_sha256"],
|
|
"profile": {k: v for k, v in manifest["input_digests"].items() if k.startswith("profile/")},
|
|
"runner": digest(__file__),
|
|
"execution_code": digest(Path(__file__).with_name("research_execution.py")),
|
|
"screen_code": digest(Path(__file__).with_name("research_admission.py")),
|
|
"workflows": load_infospace(root).config.workflows,
|
|
}, reuse_from)
|
|
elif reuse_from:
|
|
raise ValueError("Reuse is available only for conditional-v1 workspaces")
|
|
receipt.write_text(json.dumps(manifest, indent=2) + "\n")
|
|
try:
|
|
result = run_workflow(root, WORKFLOW, assisted_adapter=execution or 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__)
|
|
if isinstance(exc, InfospaceError):
|
|
manifest["error_code"] = exc.code
|
|
# Persist only the status code, never provider response bodies or headers.
|
|
status = re.search(r"HTTP Error (\d{3})", str(exc.detail.get("error", "")))
|
|
if status:
|
|
manifest["http_status"] = int(status.group(1))
|
|
raise
|
|
finally:
|
|
manifest["provider_calls_attempted"] = limited.calls
|
|
if execution:
|
|
cache = root / "reports/stage-cache.json"
|
|
cache.write_text(json.dumps(execution.cache, indent=2) + "\n")
|
|
manifest["stage_cache_sha256"] = digest(cache)
|
|
manifest["execution_events"] = execution.events
|
|
manifest["cache_hits"] = sum(e["execution"] == "cache" for e in execution.events)
|
|
manifest["deterministic_stages"] = sum(e["execution"] == "deterministic" for e in execution.events)
|
|
manifest["rejected_assertions"] = sum(e.get("rejected_assertions", 0) for e in execution.events)
|
|
manifest["editorial_corrections"] = None # Unknown until a named review exists.
|
|
if hasattr(execution, "rejected"):
|
|
(root / "reports/rejected-stage.json").write_text(json.dumps(execution.rejected, indent=2) + "\n")
|
|
if isinstance(adapter, LLMConnectAdapter):
|
|
manifest["service_requests_attempted"] = limited.calls
|
|
manifest["responses"] = adapter.responses
|
|
manifest["reported_tokens_this_run"] = sum((r.get("usage") or {}).get("total_tokens", 0)
|
|
for r in adapter.responses)
|
|
if adapter.rejected_content is not None:
|
|
rejected = root / "reports/rejected-response.md"
|
|
rejected.write_text(adapter.rejected_content)
|
|
manifest["rejected_response"] = {"path": str(rejected.relative_to(root)),
|
|
"sha256": digest(rejected),
|
|
"admission": "rejected_incomplete"}
|
|
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("--llm-connect-url", help="llm-connect service base URL")
|
|
run.add_argument("--model", help="Model or runtime profile requested from llm-connect")
|
|
run.add_argument("--max-tokens", type=int, default=4000)
|
|
run.add_argument("--diagnostics", action="store_true",
|
|
help="Retain only allowlisted effective settings and token counts")
|
|
run.add_argument("--reasoning-effort", choices=["low", "medium", "high"],
|
|
help="Explicit service model_params override; provider support may vary")
|
|
run.add_argument("--reuse-from", type=Path, help="Explicitly reuse matching completed stage results")
|
|
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 = LLMConnectAdapter(args.llm_connect_url, args.model, args.max_tokens,
|
|
args.reasoning_effort, args.diagnostics)
|
|
result = execute(args.root, adapter, args.max_calls, args.reuse_from)
|
|
print(json.dumps(result, indent=2, default=str))
|
|
except (ValueError, OSError, InfospaceError) as exc:
|
|
parser.exit(1, f"{exc}\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|