coulomb/frontend-patterns
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reduce research calls with screened reuse and draft VC-02 walkthrough

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0cffe-206f-7323-89d3-b20ca4cf4bf9
This commit is contained in:
tegwick 2026-09-25 08:43:32 +02:00
parent ad6f8fa600
commit cfa249ab13
49 changed files with 3284 additions and 207 deletions
Download patch file Download diff file Expand all files Collapse all files

49
scripts/research_bench.py
View file

@ -16,6 +16,7 @@ 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"
@ -81,7 +82,10 @@ def plan(root):
"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),
"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"}
@ -171,10 +175,12 @@ class LimitedAdapter:
return self.adapter.generate(request)
def execute(root, adapter, max_calls=4):
def execute(root, adapter, max_calls=4, reuse_from=None):
root = Path(root).resolve()
manifest = plan(root)
if manifest["provider_calls"] > max_calls:
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():
@ -183,7 +189,7 @@ def execute(root, adapter, max_calls=4):
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"]
root / "profile/workflow.yaml", *sorted((root / "profile").glob("*.json"))]
}
manifest["status"] = "started"
manifest["adapter"] = type(adapter).__name__
@ -193,10 +199,25 @@ def execute(root, adapter, max_calls=4):
"config": adapter.config.to_dict(),
"diagnostics_requested": adapter.diagnostics,
"call_limit_scope": "service requests; server retries unknown"}
receipt.write_text(json.dumps(manifest, indent=2) + "\n")
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=limited)
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__)
@ -209,9 +230,22 @@ def execute(root, adapter, max_calls=4):
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)
@ -242,6 +276,7 @@ def main():
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:
@ -259,7 +294,7 @@ def main():
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)
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")

82
scripts/research_execution.py Normal file
View file

@ -0,0 +1,82 @@
"""Conditional execution and explicit reuse of structurally screened stage results."""
import hashlib
import json
from pathlib import Path
from infospace_bench.workflow import AssistedGenerationResult, FixtureAssistedGenerationAdapter
from research_admission import json_block, screen
def fingerprint(value):
return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
def identity(adapter):
if isinstance(adapter, FixtureAssistedGenerationAdapter):
return {'fixture': sorted((list(k), v.markdown, v.provider, v.metadata)
for k, v in adapter.responses.items())}
if hasattr(adapter, 'config') and hasattr(adapter, 'endpoint'):
return {'class': type(adapter).__name__, 'endpoint': adapter.endpoint,
'config': adapter.config.to_dict(), 'diagnostics': adapter.diagnostics}
raise ValueError('Optimized execution requires llm-connect or a fixture adapter')
class ResearchExecution:
def __init__(self, limited, context, reuse_from=None):
self.limited = limited
self.context = {**context, 'adapter': identity(limited.adapter)}
self.candidates = None
self.relations = '```json\n[]\n```'
self.events, self.cache, self.previous = [], {}, {}
self.reuse_from = str(Path(reuse_from).resolve()) if reuse_from else None
if reuse_from:
root = Path(reuse_from)
receipt = json.loads((root / 'reports/research-run.json').read_text())
raw = (root / 'reports/stage-cache.json').read_bytes()
if (receipt['status'] != 'completed' or
hashlib.sha256(raw).hexdigest() != receipt.get('stage_cache_sha256')):
raise ValueError('Reuse requires a completed run and matching cache digest')
self.previous = json.loads(raw)
def generate(self, request):
stage = request.stage_id
if stage == 'relations' and len(json_block(self.candidates)['candidates']) < 2:
self.events.append({'stage': stage, 'execution': 'deterministic',
'reason': 'fewer than two candidates'})
return AssistedGenerationResult(self.relations, provider='deterministic',
metadata={'reason': 'fewer than two candidates'})
if stage == 'critique':
report = screen(self.candidates, self.relations)
self.events.append({'stage': stage, 'execution': 'deterministic', 'screen': report})
return AssistedGenerationResult(
'# Editorial review required\n\nThis is a structural screen, not model critique.\n\n'
+ '```json\n' + json.dumps(report, indent=2) + '\n```\n\n'
+ 'Check source fidelity, known-use independence, unsupported guarantees, '
'counterexamples and scenario relevance. Record corrections in the editorial review.\n',
provider='deterministic', metadata={'admission': 'not_granted'})
key = fingerprint({'context': self.context, 'stage': stage, 'prompt': request.prompt,
'workflow': request.workflow_id, 'input': request.input_artifact_id})
cached = self.previous.get(key)
event = {'stage': stage, 'execution': 'cache' if cached else 'provider'}
self.events.append(event)
if cached:
result = AssistedGenerationResult(cached['markdown'], cached['provider'],
{**cached['metadata'], 'reused_from': self.reuse_from, 'current_provider_call': False})
else:
result = self.limited.generate(request)
if stage in ('candidates', 'relations'):
report = screen(result.markdown if stage == 'candidates' else self.candidates,
'```json\n[]\n```' if stage == 'candidates' else result.markdown)
event['screen'] = report
if report['structural_status'] != 'passed':
event['rejected_assertions'] = len(report['findings'])
# Keep rejected research output for review, but do not propagate/cache it.
self.rejected = {'stage': stage, 'content': result.markdown, 'screen': report}
raise ValueError(f'{stage} structural screen blocked; inspect rejected-stage.json')
if stage == 'candidates':
self.candidates = result.markdown
if stage == 'relations':
self.relations = result.markdown
self.cache[key] = cached or {'markdown': result.markdown, 'provider': result.provider,
'metadata': result.metadata}
return result