mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-08-17 08:30:36 +00:00
allaun
cfb83cf038
sidon_weight_bound: corrected RHS from invalid sigma7(2N)/120 to
sum_{s=2}^{2N} convolutionRHS(s). Proof uses Finset.single_le_sum +
Sidon injectivity (Finset.sum_image) + E8 convolution identity.
All 5 E8 Sidon theorems now closed (0 sorries in §§9-13).
New: deepseek_v4_flash_lean_harness.py — sorry-resolution harness
targeting local llama.cpp DeepSeek V4 Flash endpoint. Scans .lean
files, sends theorem context to LLM, inserts generated proofs,
verifies with lake build, emits receipts.
Build: 3583 jobs, 0 errors (lake build)
553 lines
19 KiB
Python
553 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
DeepSeek V4 Flash Lean Harness — accelerate sorry-resolution via local llama.cpp.
|
|
|
|
Discovers `sorry` markers in a .lean file, sends each theorem (with context)
|
|
to DeepSeek V4 Flash, inserts generated proofs, and verifies with `lake build`.
|
|
|
|
Targets the local llama.cpp server at http://100.88.57.96:30516/v1
|
|
(model: deepseek-v4-flash, ~131k context).
|
|
|
|
Usage:
|
|
# Resolve all sorries in a file (iterative, one at a time)
|
|
python3 deepseek_v4_flash_lean_harness.py resolve Semantics/E8Sidon.lean
|
|
|
|
# Resolve a specific sorry by line number
|
|
python3 deepseek_v4_flash_lean_harness.py resolve Semantics/E8Sidon.lean --line 950
|
|
|
|
# List sorries without resolving
|
|
python3 deepseek_v4_flash_lean_harness.py scan Semantics/E8Sidon.lean
|
|
|
|
# Interactive mode — show each sorry, ask before sending to API
|
|
python3 deepseek_v4_flash_lean_harness.py resolve --interactive Semantics/E8Sidon.lean
|
|
|
|
Environment:
|
|
DEEPSEEK_API_BASE — defaults to http://100.88.57.96:30516/v1
|
|
DEEPSEEK_API_KEY — defaults to "sk-local"
|
|
LAKE_WORKDIR — defaults to 0-Core-Formalism/lean/Semantics
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DEFAULT_API_BASE = "http://100.88.57.96:30516/v1"
|
|
DEFAULT_API_KEY = "sk-local"
|
|
DEFAULT_MODEL = "deepseek-v4-flash"
|
|
|
|
RECEIPT_DIR = Path(__file__).resolve().parents[3] / "shared-data" / "artifacts" / "deepseek_prover"
|
|
|
|
PROMPT_TEMPLATE = """You are a Lean 4 theorem prover for the Research Stack project.
|
|
|
|
Project rules:
|
|
- Use Q16_16 fixed-point (no Float in compute paths).
|
|
- No bare sorries, no tautologies.
|
|
- Use `calc`, `omega`, `native_decide`, `positivity`, `linarith`, `nlinarith`.
|
|
- Prefer explicit `calc` blocks over opaque tactic scripts.
|
|
- Follow existing patterns in the file.
|
|
|
|
Below is a Lean 4 module with one unproven theorem (marked `:= by\n sorry`).
|
|
The imports and surrounding definitions are shown for context.
|
|
|
|
Output ONLY the proof block — the code that replaces `:= by\n sorry`.
|
|
Do NOT repeat the theorem statement. Do NOT wrap in markdown fences.
|
|
Start with `:= by` and end with the closing of the proof.
|
|
|
|
---
|
|
{context}
|
|
---
|
|
|
|
The unproven theorem at line {line_no}:
|
|
|
|
{theorem_block}
|
|
|
|
Generate the proof:"""
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class SorrySite:
|
|
line: int
|
|
theorem_name: str
|
|
theorem_block: str # from `theorem ... :=` to the `:= by\n sorry`
|
|
context_before: str
|
|
context_after: str
|
|
full_context: str
|
|
|
|
|
|
@dataclass
|
|
class HarnessConfig:
|
|
api_base: str = DEFAULT_API_BASE
|
|
api_key: str = DEFAULT_API_KEY
|
|
model: str = DEFAULT_MODEL
|
|
lake_workdir: Optional[str] = None
|
|
temperature: float = 0.4
|
|
max_tokens: int = 4096
|
|
max_iterations: int = 5
|
|
interactive: bool = False
|
|
dry_run: bool = False
|
|
|
|
|
|
@dataclass
|
|
class ProofAttempt:
|
|
sorry_site: SorrySite
|
|
code: str = ""
|
|
passed: bool = False
|
|
iterations: int = 0
|
|
compile_log: str = ""
|
|
latency_ms: float = 0.0
|
|
error_feedback: str = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sorry discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def discover_sorries(lean_path: Path) -> list[SorrySite]:
|
|
"""Scan a .lean file for `:= by\n sorry` patterns."""
|
|
text = lean_path.read_text()
|
|
lines = text.split("\n")
|
|
|
|
# Regex: find `theorem ... :=` then subsequent `sorry`
|
|
sorry_sites = []
|
|
theorem_start = None
|
|
theorem_name = None
|
|
|
|
for i, line in enumerate(lines, 1):
|
|
# Detect theorem/lemma start
|
|
m = re.match(r"^(theorem|lemma)\s+(\w+)", line)
|
|
if m:
|
|
theorem_start = i
|
|
theorem_name = m.group(2)
|
|
|
|
# Detect `:= by` or `:=` on this or next line
|
|
if theorem_start and ":=" in line and "sorry" not in line:
|
|
# Check next lines for `sorry` as a Lean keyword (not in comments)
|
|
for j in range(i, min(i + 5, len(lines) + 1)):
|
|
if j <= len(lines):
|
|
lj = lines[j - 1]
|
|
# Skip comment lines
|
|
if lj.strip().startswith("--") or lj.strip().startswith("/-") or lj.strip().startswith("*"):
|
|
continue
|
|
sorry_match = re.search(r"(?<!\w)sorry(?!\w)", lj)
|
|
if sorry_match and not lj.strip().startswith("--"):
|
|
ctx_start = max(0, theorem_start - 15)
|
|
ctx_end = min(len(lines), j + 5)
|
|
context_before = "\n".join(lines[ctx_start - 1:theorem_start - 1])
|
|
theorem_block = "\n".join(lines[theorem_start - 1:j])
|
|
context_after = "\n".join(lines[j:ctx_end])
|
|
full_context = "\n".join(lines[max(0, theorem_start - 30):min(len(lines), j + 10)])
|
|
|
|
sorry_sites.append(SorrySite(
|
|
line=j,
|
|
theorem_name=theorem_name or "unknown",
|
|
theorem_block=theorem_block,
|
|
context_before=context_before,
|
|
context_after=context_after,
|
|
full_context=full_context,
|
|
))
|
|
theorem_start = None
|
|
theorem_name = None
|
|
break
|
|
|
|
return sorry_sites
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLM API call
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def call_llm(prompt: str, cfg: HarnessConfig) -> tuple[str, float]:
|
|
"""Send prompt to DeepSeek V4 Flash via llama.cpp OpenAI-compatible API.
|
|
|
|
Returns (response_text, latency_ms).
|
|
"""
|
|
endpoint = f"{cfg.api_base.rstrip('/')}/chat/completions"
|
|
body = json.dumps({
|
|
"model": cfg.model,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": cfg.temperature,
|
|
"max_tokens": cfg.max_tokens,
|
|
"stream": False,
|
|
}).encode()
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {cfg.api_key}",
|
|
}
|
|
|
|
t0 = time.perf_counter()
|
|
req = urllib.request.Request(endpoint, data=body, headers=headers, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=180) as resp:
|
|
data = json.loads(resp.read())
|
|
latency = (time.perf_counter() - t0) * 1000
|
|
return data["choices"][0]["message"]["content"], latency
|
|
except Exception as exc:
|
|
latency = (time.perf_counter() - t0) * 1000
|
|
return f"ERROR: {exc}", latency
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lake build
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_lake_build(workdir: str, target: str = "") -> tuple[int, str]:
|
|
"""Run `lake build [target]` and return (returncode, output)."""
|
|
cmd = ["lake", "build"]
|
|
if target:
|
|
cmd.append(target)
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=240,
|
|
cwd=workdir,
|
|
)
|
|
return result.returncode, result.stdout + "\n" + result.stderr
|
|
except subprocess.TimeoutExpired as exc:
|
|
return 1, f"TIMEOUT: {exc}"
|
|
|
|
|
|
def extract_errors(log: str) -> str:
|
|
"""Extract error lines from build log."""
|
|
lines = log.split("\n")
|
|
errors = [l for l in lines if "error:" in l or "sorry" in l]
|
|
return "\n".join(errors[:15])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Proof insertion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def extract_proof_code(response: str) -> str:
|
|
"""Strip markdown fences, trim to just the Lean proof block."""
|
|
# Remove markdown fences
|
|
text = re.sub(r"^```(?:lean)?\s*\n?", "", response, flags=re.MULTILINE)
|
|
text = re.sub(r"\n```\s*$", "", text, flags=re.MULTILINE)
|
|
text = text.strip()
|
|
|
|
# If it starts with `:= by`, keep only up to the closing
|
|
if text.startswith(":= by"):
|
|
return text
|
|
|
|
# If it contains `:= by`, extract from there
|
|
idx = text.find(":= by")
|
|
if idx >= 0:
|
|
return text[idx:]
|
|
|
|
# If it contains `by` (bare proof block), extract from there
|
|
idx = text.find("\nby ")
|
|
if idx >= 0:
|
|
# Find the preceding theorem line
|
|
prev_newline = text.rfind("\n", 0, idx)
|
|
return text[prev_newline + 1:].strip()
|
|
|
|
return text
|
|
|
|
|
|
def insert_proof(lean_path: Path, sorry_line: int, proof_code: str) -> bool:
|
|
"""Replace `:= by\n sorry` at the given line with the generated proof.
|
|
|
|
Returns True if insertion succeeded.
|
|
"""
|
|
lines = lean_path.read_text().split("\n")
|
|
|
|
# Find the `:= by\n sorry` pattern starting at sorry_line
|
|
# We look for `:= by` somewhere before sorry_line, with `sorry` at sorry_line
|
|
insert_idx = None
|
|
for i in range(sorry_line - 3, sorry_line):
|
|
if i >= 0 and i < len(lines) and ":= by" in lines[i]:
|
|
insert_idx = i
|
|
break
|
|
|
|
if insert_idx is None:
|
|
# Look for `:=` on same line as `sorry`
|
|
if sorry_line - 1 < len(lines) and ":=" in lines[sorry_line - 1] and "sorry" in lines[sorry_line - 1]:
|
|
insert_idx = sorry_line - 1
|
|
|
|
if insert_idx is None:
|
|
print(f" No `:= by` found before line {sorry_line}")
|
|
return False
|
|
|
|
# Replace from `:=` onwards with the proof
|
|
indent = " " # 2-space indent matching project style
|
|
proof_lines = proof_code.split("\n")
|
|
if len(proof_lines) == 1:
|
|
# Single line: replace `:= by\n sorry` with proof_code
|
|
# Remove `:= by` at insert_idx and `sorry` at sorry_line
|
|
header = lines[insert_idx].split(":= by")[0].rstrip()
|
|
new_lines = lines[:insert_idx] + [f"{header} {proof_code}"] + lines[sorry_line:]
|
|
else:
|
|
# Multi-line proof
|
|
header = lines[insert_idx].split(":= by")[0].rstrip()
|
|
# Keep `:= by` header, replace the sorry line(s) with proof body
|
|
proof_body = "\n".join(
|
|
f"{indent}{l}" if l.strip() and not l.startswith(indent) else l
|
|
for l in proof_lines[1:] if not l.startswith(":= by")
|
|
)
|
|
# Count how many sorry lines to remove
|
|
sorry_count = 1
|
|
for j in range(sorry_line, min(sorry_line + 3, len(lines))):
|
|
if "sorry" in lines[j - 1] or lines[j - 1].strip() == "":
|
|
sorry_count = j - sorry_line + 1
|
|
else:
|
|
break
|
|
|
|
new_lines = (
|
|
lines[:insert_idx]
|
|
+ [f"{header} := by"]
|
|
+ [proof_body]
|
|
+ lines[sorry_line + sorry_count - 1:]
|
|
)
|
|
|
|
lean_path.write_text("\n".join(new_lines))
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Receipt emission
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def emit_receipt(attempt: ProofAttempt, cfg: HarnessConfig) -> Path:
|
|
"""Write a proof attempt receipt."""
|
|
RECEIPT_DIR.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
safe_name = attempt.sorry_site.theorem_name[:40]
|
|
fname = f"proof_attempt_{safe_name}_{ts}.json"
|
|
|
|
receipt = {
|
|
"schema": "deepseek_v4_flash_proof_attempt_v1",
|
|
"model": cfg.model,
|
|
"theorem": attempt.sorry_site.theorem_name,
|
|
"line": attempt.sorry_site.line,
|
|
"passed": attempt.passed,
|
|
"iterations": attempt.iterations,
|
|
"latency_ms": attempt.latency_ms,
|
|
"timestamp": ts,
|
|
"error_preview": extract_errors(attempt.compile_log)[:500] if not attempt.passed else "",
|
|
}
|
|
path = RECEIPT_DIR / fname
|
|
path.write_text(json.dumps(receipt, indent=2) + "\n")
|
|
return path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main resolve loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def resolve_sorry(site: SorrySite, cfg: HarnessConfig, lean_path: Path) -> ProofAttempt:
|
|
"""Attempt to resolve a single sorry site."""
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Theorem: {site.theorem_name} (line {site.line})")
|
|
print(f"{'=' * 60}")
|
|
print(site.theorem_block[:200] + "..." if len(site.theorem_block) > 200 else site.theorem_block)
|
|
|
|
if cfg.interactive:
|
|
resp = input("\nSend to DeepSeek V4 Flash? [Y/n] ").strip().lower()
|
|
if resp == "n":
|
|
print("Skipping.")
|
|
return ProofAttempt(sorry_site=site, passed=False, iterations=0)
|
|
|
|
attempt = ProofAttempt(sorry_site=site)
|
|
|
|
for iteration in range(1, cfg.max_iterations + 1):
|
|
print(f"\n--- Iteration {iteration}/{cfg.max_iterations} ---")
|
|
|
|
# Build prompt
|
|
context = site.full_context
|
|
error_feedback = attempt.error_feedback
|
|
if error_feedback:
|
|
prompt = PROMPT_TEMPLATE + f"\n\nPrevious attempt failed. Errors:\n{error_feedback}\n\nTry a different approach:"
|
|
else:
|
|
prompt = PROMPT_TEMPLATE.format(
|
|
context=context,
|
|
line_no=site.line,
|
|
theorem_block=site.theorem_block,
|
|
)
|
|
|
|
if cfg.dry_run:
|
|
print(f"\n[DRY RUN] Would send prompt ({len(prompt)} chars)")
|
|
print(f"--- prompt preview ---\n{prompt[:500]}...\n---")
|
|
continue
|
|
|
|
# Call LLM
|
|
response, latency = call_llm(prompt, cfg)
|
|
attempt.latency_ms += latency
|
|
print(f" API: {latency:.0f}ms")
|
|
|
|
if response.startswith("ERROR:"):
|
|
print(f" {response}")
|
|
if iteration < cfg.max_iterations:
|
|
continue
|
|
break
|
|
|
|
# Extract proof code
|
|
proof_code = extract_proof_code(response)
|
|
print(f" Generated: {len(proof_code)} chars")
|
|
if not proof_code:
|
|
print(" Empty response, retrying...")
|
|
continue
|
|
|
|
# Insert into file
|
|
if not insert_proof(lean_path, site.line, proof_code):
|
|
print(" Failed to insert proof")
|
|
continue
|
|
|
|
# Build
|
|
workdir = cfg.lake_workdir or os.environ.get("LAKE_WORKDIR", "")
|
|
rc, log = run_lake_build(workdir)
|
|
attempt.compile_log = log
|
|
attempt.code = proof_code
|
|
attempt.iterations = iteration
|
|
|
|
if rc == 0:
|
|
print(f" \033[32mPASSED!\033[0m (iteration {iteration})")
|
|
attempt.passed = True
|
|
return attempt
|
|
|
|
# Extract errors for feedback
|
|
errors = extract_errors(log)
|
|
attempt.error_feedback = errors[:2000]
|
|
print(f" \033[31mFAILED\033[0m (return code {rc})")
|
|
if errors:
|
|
print(f" Errors: {errors[:300]}...")
|
|
|
|
# Revert the insertion for next iteration
|
|
# Read current state, check if the proof was added
|
|
# If it failed, the file has the broken proof now; we need to restore sorry
|
|
if not insert_proof(lean_path, site.line, " sorry"):
|
|
print(" Warning: could not restore sorry marker")
|
|
|
|
return attempt
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def cmd_scan(args):
|
|
"""Scan a file for sorries and print them."""
|
|
path = Path(args.lean_file)
|
|
if not path.exists():
|
|
print(f"File not found: {path}")
|
|
sys.exit(1)
|
|
|
|
sites = discover_sorries(path)
|
|
if not sites:
|
|
print("No sorries found.")
|
|
return
|
|
|
|
print(f"Found {len(sites)} sorry site(s) in {path}:")
|
|
for s in sites:
|
|
print(f" Line {s.line:>5}: {s.theorem_name}")
|
|
|
|
|
|
def cmd_resolve(args):
|
|
"""Resolve sorries in a file."""
|
|
path = Path(args.lean_file)
|
|
if not path.exists():
|
|
print(f"File not found: {path}")
|
|
sys.exit(1)
|
|
|
|
cfg = HarnessConfig(
|
|
api_base=args.api_base or os.environ.get("DEEPSEEK_API_BASE", DEFAULT_API_BASE),
|
|
api_key=args.api_key or os.environ.get("DEEPSEEK_API_KEY", DEFAULT_API_KEY),
|
|
model=args.model or DEFAULT_MODEL,
|
|
lake_workdir=args.lake_workdir or os.environ.get("LAKE_WORKDIR", ""),
|
|
temperature=args.temperature,
|
|
max_iterations=args.max_iterations,
|
|
interactive=args.interactive,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
sites = discover_sorries(path)
|
|
if args.line:
|
|
sites = [s for s in sites if s.line == args.line]
|
|
if not sites:
|
|
print(f"No sorry at line {args.line}")
|
|
sys.exit(1)
|
|
|
|
if not sites:
|
|
print("No sorries found.")
|
|
return
|
|
|
|
print(f"Found {len(sites)} sorry site(s).")
|
|
passed = 0
|
|
failed = 0
|
|
|
|
for site in sites:
|
|
attempt = resolve_sorry(site, cfg, path)
|
|
if attempt.passed:
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
|
|
# Emit receipt
|
|
receipt_path = emit_receipt(attempt, cfg)
|
|
print(f" Receipt: {receipt_path}")
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Results: {passed} passed, {failed} failed, {len(sites)} total")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="DeepSeek V4 Flash Lean Harness — accelerate sorry resolution",
|
|
)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# scan
|
|
scan_p = sub.add_parser("scan", help="List sorries in a file")
|
|
scan_p.add_argument("lean_file", help="Path to .lean file")
|
|
|
|
# resolve
|
|
res_p = sub.add_parser("resolve", help="Resolve sorries in a file")
|
|
res_p.add_argument("lean_file", help="Path to .lean file")
|
|
res_p.add_argument("--line", type=int, default=0, help="Specific sorry line to resolve")
|
|
res_p.add_argument("--interactive", "-i", action="store_true", help="Ask before each API call")
|
|
res_p.add_argument("--dry-run", "-n", action="store_true", help="Show prompts without sending")
|
|
res_p.add_argument("--max-iterations", type=int, default=5, help="Max generate-compile cycles per sorry")
|
|
res_p.add_argument("--temperature", type=float, default=0.4, help="LLM temperature (default 0.4)")
|
|
res_p.add_argument("--model", default="", help="Model name (default deepseek-v4-flash)")
|
|
res_p.add_argument("--api-base", default="", help=f"API base URL (default {DEFAULT_API_BASE})")
|
|
res_p.add_argument("--api-key", default="", help="API key (default sk-local)")
|
|
res_p.add_argument("--lake-workdir", default="", help="lake build working directory")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "scan":
|
|
cmd_scan(args)
|
|
elif args.command == "resolve":
|
|
cmd_resolve(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|