9726042964
Committed code predated the line-length=100 ruff config; this brings the five drifted files into compliance. No logic change. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
859 lines
29 KiB
Python
859 lines
29 KiB
Python
"""Synthesis / Lead-Engine (F-13).
|
|
|
|
Generates four kinds of research **leads** over the RAG substrate:
|
|
|
|
| Stage | Kind | Grounded? | Output dir |
|
|
|-------|---------------|-----------|--------------------------|
|
|
| 2 | connection | yes | ``leads/grounded/`` |
|
|
| 3 | gap | yes | ``leads/grounded/`` |
|
|
| 4 | improvement | yes | ``leads/grounded/`` |
|
|
| 5 | conjecture | no — by definition unverifiable | ``leads/conjectures/`` |
|
|
|
|
Hard invariant (F-13 HARD)
|
|
--------------------------
|
|
**Conjectures are NEVER written to ``wiki/``.** They are quarantined to
|
|
``leads/conjectures/`` with mandatory ``status="unverified"`` plus a
|
|
``suggested_validation`` field describing the spike that would falsify them.
|
|
|
|
The wiki layer (F-12) stays grounded-only. Conjectures live in their own
|
|
sandbox so that human review can promote them after a spike succeeds.
|
|
|
|
Grounding discipline
|
|
--------------------
|
|
* For ``connection`` / ``gap`` / ``improvement`` leads we run the
|
|
same content-5-gram grounding check as F-12 (re-used directly from
|
|
:mod:`codex.wiki`). Every claim in the lead body that fails the check
|
|
is marked with the ⚠ prefix and the lead is recorded as having
|
|
ungrounded claims. Leads whose grounded-ratio is below the floor are
|
|
dropped (returned as "ungrounded" rather than "grounded").
|
|
* For ``conjecture`` leads grounding by definition cannot apply
|
|
(the claim does not exist in the corpus). Provenance is still mandatory:
|
|
it points at the chunks that *seeded* the conjecture so a reviewer can
|
|
trace the inspiration.
|
|
|
|
Graceful degradation
|
|
--------------------
|
|
* Missing ``lib_path`` → :func:`find_improvements` returns ``[]``.
|
|
* DB unreachable → caller sees the psycopg error; no silent failures.
|
|
* LLM unreachable (httpx.ConnectError / any exception) → the affected
|
|
generator returns ``[]`` (matches the F-12 LLM degradation rule).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any, Literal, Protocol
|
|
|
|
from codex.config import get_settings
|
|
from codex.wiki import (
|
|
Claim,
|
|
LLMClient,
|
|
OllamaClient,
|
|
_parse_claims,
|
|
_retrieve_chunks,
|
|
_run_grounding_guard,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Datamodel
|
|
# ---------------------------------------------------------------------------
|
|
|
|
LeadKind = Literal["connection", "gap", "improvement", "conjecture"]
|
|
LeadStatus = Literal["unverified", "spiking", "go", "no-go"]
|
|
|
|
|
|
@dataclass
|
|
class Provenance:
|
|
"""Pointer back to a single source location for a lead.
|
|
|
|
``locator`` follows the same conventions as F-12 wiki claims:
|
|
``"chunk 16"``, ``"page 9"``, ``"eq.(9)"``, or — for ``improvement``
|
|
leads — a code locator like ``"src/cli.cpp:42"``.
|
|
"""
|
|
|
|
bibkey: str
|
|
locator: str
|
|
|
|
|
|
@dataclass
|
|
class Lead:
|
|
"""A single research lead emitted by the synthesis engine.
|
|
|
|
See module docstring for the four-stage taxonomy and the
|
|
grounded-vs-conjecture invariant.
|
|
"""
|
|
|
|
id: str # "L-0001"
|
|
kind: LeadKind
|
|
title: str
|
|
body: str
|
|
provenance: list[Provenance] # mandatory — even for conjectures
|
|
confidence: float # 0..1
|
|
status: LeadStatus = "unverified"
|
|
# For conjectures: the falsification spike. Empty string for grounded leads.
|
|
suggested_validation: str = ""
|
|
# Derived: claim-level grounding decisions (empty for conjectures).
|
|
claims: list[Claim] = field(default_factory=list)
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.provenance:
|
|
raise ValueError(
|
|
f"Lead '{self.id}' must carry at least one Provenance entry. "
|
|
"For gap leads with no chunks, use a topic-marker Provenance."
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Protocols
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _ConnLike(Protocol):
|
|
"""Subset of ``psycopg.Connection`` used by this module (helps tests)."""
|
|
|
|
def execute(self, sql: str, params: dict[str, Any] | None = ...) -> Any: ...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLM prompts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CONNECTION_PROMPT = """\
|
|
You are a research analyst surfacing CONNECTIONS between papers.
|
|
|
|
A connection is a non-trivial overlap (shared definitions, shared techniques,
|
|
shared theorems, or one paper extending/generalising another). It is a
|
|
DESCRIPTIVE statement about the corpus — not a conjecture.
|
|
|
|
Use ONLY the source chunks below. For every factual claim, cite it inline
|
|
in the format [BibKey #locator]. Example:
|
|
"Both papers use the Lobachevsky function as the building block of volume
|
|
formulas. [Springborn2008 #chunk 16] [BobenkoPinkallSpringborn2015 #chunk 3]"
|
|
|
|
Do NOT invent facts. Do NOT speculate. Two short paragraphs maximum.
|
|
|
|
SOURCE CHUNKS:
|
|
{chunks_block}
|
|
|
|
Now describe a single non-trivial connection between these papers:
|
|
"""
|
|
|
|
_GAP_PROMPT = """\
|
|
You are a research analyst identifying GAPS in a corpus.
|
|
|
|
A gap is an ABSENCE: a topic, technique, or comparison that the corpus
|
|
fails to cover, even though adjacent material exists. Make the absence
|
|
concrete by pointing at where the corpus DOES talk about adjacent things.
|
|
|
|
Use ONLY the source chunks below. For every factual claim about what IS
|
|
in the corpus, cite it inline in the format [BibKey #locator].
|
|
The absence itself does not need a citation; the surrounding context does.
|
|
|
|
Do NOT invent facts. Two short paragraphs maximum.
|
|
|
|
SOURCE CHUNKS:
|
|
{chunks_block}
|
|
|
|
Now describe a single concrete gap in the corpus's coverage of "{topic}":
|
|
"""
|
|
|
|
_IMPROVEMENT_PROMPT = """\
|
|
You are a research analyst inspecting C++ code annotated with @cite tags.
|
|
|
|
For the symbol below, a literature paper is cited. Read the chunks and
|
|
suggest at most one CONCRETE improvement to the code that the literature
|
|
would support. The improvement must be a DESCRIPTIVE statement of what the
|
|
paper says, not a conjecture.
|
|
|
|
Use ONLY the source chunks. Cite each claim inline as [BibKey #locator].
|
|
|
|
Symbol: {symbol}
|
|
Cited bibkey: {bibkey}
|
|
File:locator: {file_locator}
|
|
|
|
SOURCE CHUNKS:
|
|
{chunks_block}
|
|
|
|
Now state a single grounded improvement (or "No improvement suggested." if
|
|
the chunks do not support any):
|
|
"""
|
|
|
|
_CONJECTURE_PROMPT = """\
|
|
You are a creative researcher proposing CONJECTURES — novel hypotheses that
|
|
EXTEND but DO NOT contradict the corpus.
|
|
|
|
A conjecture is a NEW idea not yet present in the literature. By definition
|
|
it cannot be grounded in the chunks. Your output MUST be honest about this:
|
|
do not pretend it is a result already in the papers.
|
|
|
|
You MUST produce:
|
|
1. A one-sentence title (the conjecture itself).
|
|
2. A short body (2-4 sentences) explaining the intuition.
|
|
3. A "Validation:" line stating ONE concrete spike — a small experiment,
|
|
counter-example search, or proof attempt — that would FALSIFY the
|
|
conjecture if it is wrong. Without a falsification path the conjecture
|
|
is rejected.
|
|
|
|
Use the chunks below ONLY as inspiration. Cite seed material as
|
|
[BibKey #locator] in the body so a reviewer can trace where the idea came
|
|
from.
|
|
|
|
SOURCE CHUNKS:
|
|
{chunks_block}
|
|
|
|
Now propose one conjecture in the format:
|
|
Title: <one sentence>
|
|
Body: <2-4 sentences with [BibKey #locator] seed citations>
|
|
Validation: <one concrete falsification spike>
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _format_chunks(chunks: list[dict[str, Any]]) -> str:
|
|
"""Render chunks for the LLM prompt — same format as F-12."""
|
|
lines: list[str] = []
|
|
for chunk in chunks:
|
|
bibkey = chunk.get("bibkey") or chunk["paper_id"]
|
|
ord_val = chunk.get("ord", "?")
|
|
lines.append(f"[{bibkey} #chunk {ord_val}]\n{chunk['content'].strip()}\n")
|
|
return "\n---\n".join(lines)
|
|
|
|
|
|
def _mark_ungrounded(body: str, claims: list[Claim]) -> str:
|
|
"""Prefix each ungrounded claim's last sentence with ⚠ in *body*."""
|
|
for claim in claims:
|
|
if claim.grounded:
|
|
continue
|
|
escaped = re.escape(claim.text.strip())
|
|
body = re.sub(rf"(?<!⚠ )({escaped})", r"⚠ \1", body, count=1)
|
|
return body
|
|
|
|
|
|
def _grounded_ratio(claims: list[Claim]) -> float:
|
|
"""Fraction of *claims* that passed the grounding check (0.0 if empty)."""
|
|
if not claims:
|
|
return 0.0
|
|
n_grounded = sum(1 for c in claims if c.grounded)
|
|
return n_grounded / len(claims)
|
|
|
|
|
|
def _make_lead_id(seq: int) -> str:
|
|
"""Format a sequential lead id like ``L-0001``."""
|
|
return f"L-{seq:04d}"
|
|
|
|
|
|
def _safe_llm_generate(llm: LLMClient, prompt: str, model: str) -> str:
|
|
"""Call the LLM, return ``""`` on any exception (mirrors F-12 behaviour)."""
|
|
try:
|
|
return llm.generate(prompt, model=model)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("LLM unavailable for synthesis (%s)", exc)
|
|
return ""
|
|
|
|
|
|
def _provenance_from_chunks(chunks: list[dict[str, Any]]) -> list[Provenance]:
|
|
"""Build Provenance pointers for every chunk used as a seed."""
|
|
out: list[Provenance] = []
|
|
for chunk in chunks:
|
|
bibkey = str(chunk.get("bibkey") or chunk.get("paper_id") or "")
|
|
ord_val = chunk.get("ord", "?")
|
|
out.append(Provenance(bibkey=bibkey, locator=f"chunk {ord_val}"))
|
|
return out
|
|
|
|
|
|
def _finalise_grounded_lead(
|
|
*,
|
|
seq: int,
|
|
kind: LeadKind,
|
|
title: str,
|
|
body: str,
|
|
chunks: list[dict[str, Any]],
|
|
min_grounded_ratio: float,
|
|
) -> Lead | None:
|
|
"""Parse claims, run grounding guard, mark ⚠, return ``None`` if too sparse.
|
|
|
|
Returns the populated :class:`Lead` or ``None`` if no parseable claims
|
|
were found or the grounded ratio is below ``min_grounded_ratio``.
|
|
"""
|
|
claims = _parse_claims(body)
|
|
if not claims:
|
|
# No citations at all → cannot be a grounded lead.
|
|
return None
|
|
claims = _run_grounding_guard(claims, chunks)
|
|
ratio = _grounded_ratio(claims)
|
|
if ratio < min_grounded_ratio:
|
|
return None
|
|
marked = _mark_ungrounded(body, claims)
|
|
return Lead(
|
|
id=_make_lead_id(seq),
|
|
kind=kind,
|
|
title=title,
|
|
body=marked,
|
|
provenance=_provenance_from_chunks(chunks),
|
|
confidence=ratio,
|
|
status="unverified",
|
|
suggested_validation="",
|
|
claims=claims,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 2 — connections
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def find_connections(
|
|
*,
|
|
db_conn: _ConnLike | None = None,
|
|
top_k: int,
|
|
llm: LLMClient,
|
|
) -> list[Lead]:
|
|
"""Surface grounded connection leads across the corpus.
|
|
|
|
Strategy: cluster the corpus by shared retrieval — for each paper
|
|
pair that shows up together in the top-K of a common probe query
|
|
we ask the LLM to articulate the connection. The probe queries are
|
|
the bibkey/title pairs of the existing papers (cheap, no external
|
|
config needed).
|
|
|
|
Parameters
|
|
----------
|
|
db_conn:
|
|
Unused right now (kept for the agreed signature so callers can
|
|
inject a transactional connection later).
|
|
top_k:
|
|
Number of chunks retrieved per probe.
|
|
llm:
|
|
Injectable LLM client. Errors fall back to ``[]``.
|
|
"""
|
|
settings = get_settings()
|
|
seq = 1
|
|
leads: list[Lead] = []
|
|
|
|
# Probe queries: use the discovered titles of already-ingested papers.
|
|
# Without DB we can't get titles, so we fall back to a generic probe.
|
|
titles = _list_paper_titles(db_conn)
|
|
if not titles:
|
|
return leads
|
|
|
|
seen_pairs: set[tuple[str, str]] = set()
|
|
|
|
for bibkey, title in titles:
|
|
chunks = _retrieve_chunks([title], top_k=top_k)
|
|
if not chunks:
|
|
continue
|
|
# Find chunks from a *different* bibkey — that's the connection signal.
|
|
other_bibkeys = {
|
|
str(c.get("bibkey") or "")
|
|
for c in chunks
|
|
if c.get("bibkey") and c.get("bibkey") != bibkey
|
|
}
|
|
for other in sorted(other_bibkeys):
|
|
pair = tuple(sorted([bibkey, other]))
|
|
if pair in seen_pairs:
|
|
continue
|
|
seen_pairs.add(pair) # type: ignore[arg-type]
|
|
|
|
pair_chunks = [c for c in chunks if c.get("bibkey") in {bibkey, other}]
|
|
prompt = _CONNECTION_PROMPT.format(chunks_block=_format_chunks(pair_chunks))
|
|
raw = _safe_llm_generate(llm, prompt, settings.synthesis_llm_model)
|
|
if not raw.strip():
|
|
continue
|
|
lead = _finalise_grounded_lead(
|
|
seq=seq,
|
|
kind="connection",
|
|
title=f"Connection: {bibkey} ⇄ {other}",
|
|
body=raw.strip(),
|
|
chunks=pair_chunks,
|
|
min_grounded_ratio=settings.synthesis_min_grounded_ratio,
|
|
)
|
|
if lead is None:
|
|
continue
|
|
leads.append(lead)
|
|
seq += 1
|
|
|
|
return leads
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 3 — gaps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def find_gaps(
|
|
lib_path: str | None = None,
|
|
*,
|
|
db_conn: _ConnLike | None = None,
|
|
llm: LLMClient,
|
|
topics: list[str] | None = None,
|
|
) -> list[Lead]:
|
|
"""Surface grounded gap leads.
|
|
|
|
Two complementary signals are used:
|
|
|
|
* **Topic coverage map** — for each topic in ``topics`` (or, if
|
|
``None``, the union of paper titles), retrieve chunks and check
|
|
bibkey coverage. A topic covered by < ``synthesis_gap_min_coverage``
|
|
bibkeys is reported as a gap candidate and the LLM is asked to
|
|
articulate it.
|
|
* **Code-vs-corpus gap** — when ``lib_path`` is given, any @cite
|
|
bibkey scanned from code that is NOT present in the corpus is
|
|
flagged. This catches the "cited in code, never ingested" case.
|
|
"""
|
|
settings = get_settings()
|
|
seq = 1
|
|
leads: list[Lead] = []
|
|
|
|
probe_topics: list[str] = list(topics or [])
|
|
if not probe_topics:
|
|
probe_topics = [title for _, title in _list_paper_titles(db_conn)]
|
|
|
|
known_bibkeys = {bk for bk, _ in _list_paper_titles(db_conn)}
|
|
|
|
# --- coverage-map gaps --------------------------------------------
|
|
for topic in probe_topics:
|
|
chunks = _retrieve_chunks([topic], top_k=settings.synthesis_top_k)
|
|
if not chunks:
|
|
# No coverage at all → record the gap directly (LLM not strictly needed).
|
|
leads.append(
|
|
Lead(
|
|
id=_make_lead_id(seq),
|
|
kind="gap",
|
|
title=f"Gap: corpus has no material on '{topic}'",
|
|
body=(
|
|
f"The retrieval layer returned no chunks for topic '{topic}'.\n"
|
|
"No grounded statement is possible; this is the absence itself."
|
|
),
|
|
provenance=[Provenance(bibkey="(none)", locator=f"topic:{topic}")],
|
|
confidence=1.0,
|
|
status="unverified",
|
|
suggested_validation=(
|
|
"Ingest a paper covering this topic, re-run synthesis: the gap "
|
|
"should disappear."
|
|
),
|
|
)
|
|
)
|
|
seq += 1
|
|
continue
|
|
|
|
bibkeys_for_topic = {str(c.get("bibkey") or "") for c in chunks if c.get("bibkey")}
|
|
if len(bibkeys_for_topic) >= settings.synthesis_gap_min_coverage:
|
|
# Topic is well covered — not a gap.
|
|
continue
|
|
prompt = _GAP_PROMPT.format(topic=topic, chunks_block=_format_chunks(chunks))
|
|
raw = _safe_llm_generate(llm, prompt, settings.synthesis_llm_model)
|
|
if not raw.strip():
|
|
continue
|
|
lead = _finalise_grounded_lead(
|
|
seq=seq,
|
|
kind="gap",
|
|
title=f"Gap: '{topic}' covered by only {len(bibkeys_for_topic)} bibkey(s)",
|
|
body=raw.strip(),
|
|
chunks=chunks,
|
|
min_grounded_ratio=settings.synthesis_min_grounded_ratio,
|
|
)
|
|
if lead is None:
|
|
continue
|
|
leads.append(lead)
|
|
seq += 1
|
|
|
|
# --- code-vs-corpus gaps ------------------------------------------
|
|
if lib_path:
|
|
from codex.provenance import scan_cite_tags
|
|
|
|
try:
|
|
hits = scan_cite_tags(lib_path)
|
|
except FileNotFoundError:
|
|
hits = []
|
|
cited_bibkeys = {h["bibkey"] for h in hits}
|
|
missing = cited_bibkeys - known_bibkeys
|
|
for bibkey in sorted(missing):
|
|
leads.append(
|
|
Lead(
|
|
id=_make_lead_id(seq),
|
|
kind="gap",
|
|
title=f"Gap: code cites '{bibkey}' but corpus does not contain it",
|
|
body=(
|
|
f"The bibkey `{bibkey}` is referenced by @cite tags in `{lib_path}`\n"
|
|
"but no paper with this bibkey is ingested into the corpus.\n"
|
|
"This is a concrete absence."
|
|
),
|
|
provenance=[Provenance(bibkey=bibkey, locator="(not ingested)")],
|
|
confidence=1.0,
|
|
status="unverified",
|
|
suggested_validation=(
|
|
f"Ingest the paper for `{bibkey}` via `codex ingest`. The gap "
|
|
"should disappear on the next synthesis run."
|
|
),
|
|
)
|
|
)
|
|
seq += 1
|
|
|
|
return leads
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 4 — improvements
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def find_improvements(
|
|
lib_path: str,
|
|
*,
|
|
db_conn: _ConnLike | None = None,
|
|
top_k: int,
|
|
llm: LLMClient,
|
|
) -> list[Lead]:
|
|
"""Suggest grounded code improvements based on @cite-linked literature.
|
|
|
|
Walks ``lib_path`` for ``@cite <bibkey>`` markers, retrieves the
|
|
relevant chunks for each cited bibkey, and asks the LLM to propose
|
|
one concrete improvement supported by the paper. Improvements
|
|
failing the grounding check are dropped.
|
|
"""
|
|
if not lib_path:
|
|
return []
|
|
from codex.provenance import scan_cite_tags
|
|
|
|
settings = get_settings()
|
|
seq = 1
|
|
leads: list[Lead] = []
|
|
|
|
try:
|
|
hits = scan_cite_tags(lib_path)
|
|
except FileNotFoundError:
|
|
return leads
|
|
|
|
known_bibkeys = {bk for bk, _ in _list_paper_titles(db_conn)}
|
|
|
|
# Group hits by (symbol, bibkey) so we ask the LLM at most once per pair.
|
|
by_key: dict[tuple[str, str], list[dict[str, str]]] = {}
|
|
for hit in hits:
|
|
if hit["bibkey"] not in known_bibkeys:
|
|
continue # handled by find_gaps
|
|
by_key.setdefault((hit["symbol"] or "(unknown)", hit["bibkey"]), []).append(hit)
|
|
|
|
for (symbol, bibkey), group in by_key.items():
|
|
chunks = _retrieve_chunks([symbol, bibkey], top_k=top_k)
|
|
# Restrict to chunks from the cited bibkey when possible
|
|
same_bib = [c for c in chunks if c.get("bibkey") == bibkey] or chunks
|
|
if not same_bib:
|
|
continue
|
|
sample = group[0]
|
|
file_locator = f"{sample['file']}:{sample['line']}"
|
|
prompt = _IMPROVEMENT_PROMPT.format(
|
|
symbol=symbol,
|
|
bibkey=bibkey,
|
|
file_locator=file_locator,
|
|
chunks_block=_format_chunks(same_bib),
|
|
)
|
|
raw = _safe_llm_generate(llm, prompt, settings.synthesis_llm_model)
|
|
if not raw.strip() or "no improvement suggested" in raw.lower():
|
|
continue
|
|
lead = _finalise_grounded_lead(
|
|
seq=seq,
|
|
kind="improvement",
|
|
title=f"Improvement: {symbol} ({file_locator}) via {bibkey}",
|
|
body=raw.strip(),
|
|
chunks=same_bib,
|
|
min_grounded_ratio=settings.synthesis_min_grounded_ratio,
|
|
)
|
|
if lead is None:
|
|
continue
|
|
# Attach the code locator to the provenance list as the first entry
|
|
lead.provenance.insert(0, Provenance(bibkey=bibkey, locator=file_locator))
|
|
leads.append(lead)
|
|
seq += 1
|
|
|
|
return leads
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 5 — conjectures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_CONJECTURE_RE = re.compile(
|
|
r"Title:\s*(?P<title>.+?)\n+"
|
|
r"Body:\s*(?P<body>.+?)\n+"
|
|
r"Validation:\s*(?P<validation>.+)",
|
|
re.DOTALL | re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _parse_conjecture_output(raw: str) -> tuple[str, str, str] | None:
|
|
"""Parse the LLM's three-section conjecture output.
|
|
|
|
Returns ``(title, body, validation)`` or ``None`` if any section
|
|
is missing or empty.
|
|
"""
|
|
match = _CONJECTURE_RE.search(raw)
|
|
if not match:
|
|
return None
|
|
title = match.group("title").strip()
|
|
body = match.group("body").strip()
|
|
validation = match.group("validation").strip()
|
|
if not title or not body or not validation:
|
|
return None
|
|
return title, body, validation
|
|
|
|
|
|
def propose_conjectures(
|
|
*,
|
|
db_conn: _ConnLike | None = None,
|
|
top_k: int,
|
|
llm: LLMClient,
|
|
seeds: list[str] | None = None,
|
|
) -> list[Lead]:
|
|
"""Generate stage-5 conjecture leads.
|
|
|
|
Each conjecture is the LLM's own hypothesis seeded by a small
|
|
retrieval slice. The output is by definition ungrounded, so:
|
|
|
|
* ``status`` is hard-set to ``"unverified"``.
|
|
* ``suggested_validation`` is mandatory; conjectures without a
|
|
falsification path are dropped.
|
|
* ``provenance`` records the seed chunks so a reviewer can trace
|
|
inspiration.
|
|
|
|
Conjectures are written to ``leads/conjectures/`` only — never
|
|
to ``wiki/``.
|
|
"""
|
|
settings = get_settings()
|
|
seq = 1
|
|
leads: list[Lead] = []
|
|
|
|
probe_seeds: list[str] = list(seeds or [])
|
|
if not probe_seeds:
|
|
probe_seeds = [title for _, title in _list_paper_titles(db_conn)]
|
|
if not probe_seeds:
|
|
return leads
|
|
|
|
for seed in probe_seeds:
|
|
chunks = _retrieve_chunks([seed], top_k=top_k)
|
|
if not chunks:
|
|
continue
|
|
prompt = _CONJECTURE_PROMPT.format(chunks_block=_format_chunks(chunks))
|
|
raw = _safe_llm_generate(llm, prompt, settings.synthesis_llm_model)
|
|
if not raw.strip():
|
|
continue
|
|
parsed = _parse_conjecture_output(raw)
|
|
if parsed is None:
|
|
# No falsification path → reject (hard invariant).
|
|
continue
|
|
title, body, validation = parsed
|
|
provenance = _provenance_from_chunks(chunks)
|
|
if not provenance:
|
|
continue
|
|
leads.append(
|
|
Lead(
|
|
id=_make_lead_id(seq),
|
|
kind="conjecture",
|
|
title=title,
|
|
body=body,
|
|
provenance=provenance,
|
|
confidence=0.0, # conjectures start unconfirmed
|
|
status="unverified", # HARD invariant
|
|
suggested_validation=validation,
|
|
)
|
|
)
|
|
seq += 1
|
|
return leads
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _list_paper_titles(db_conn: _ConnLike | None) -> list[tuple[str, str]]:
|
|
"""Return ``[(bibkey, title), …]`` for all ingested papers with a bibkey.
|
|
|
|
Uses the injected connection when provided, otherwise opens a fresh
|
|
one. Returns ``[]`` if the database is unreachable or has no rows.
|
|
"""
|
|
sql = "SELECT bibkey, title FROM papers WHERE bibkey IS NOT NULL"
|
|
try:
|
|
if db_conn is not None:
|
|
rows = db_conn.execute(sql).fetchall()
|
|
else:
|
|
from codex.db import get_conn
|
|
|
|
with get_conn() as conn:
|
|
rows = conn.execute(sql).fetchall()
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("Could not list paper titles (%s)", exc)
|
|
return []
|
|
out: list[tuple[str, str]] = []
|
|
for row in rows:
|
|
bibkey = str(row["bibkey"]) if "bibkey" in row else str(row[0])
|
|
title = str(row["title"]) if "title" in row else str(row[1])
|
|
out.append((bibkey, title))
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_GROUNDED_KINDS: frozenset[LeadKind] = frozenset({"connection", "gap", "improvement"})
|
|
|
|
|
|
def _lead_to_markdown(lead: Lead) -> str:
|
|
"""Render a Lead as a self-contained markdown document.
|
|
|
|
Front-matter is a fenced JSON block so the file is human-readable
|
|
and machine-parseable.
|
|
"""
|
|
meta = {
|
|
"id": lead.id,
|
|
"kind": lead.kind,
|
|
"status": lead.status,
|
|
"confidence": round(lead.confidence, 3),
|
|
"provenance": [asdict(p) for p in lead.provenance],
|
|
"suggested_validation": lead.suggested_validation,
|
|
"compiled_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
parts = [
|
|
f"# {lead.id} — {lead.title}",
|
|
"",
|
|
"```json",
|
|
json.dumps(meta, indent=2, sort_keys=True),
|
|
"```",
|
|
"",
|
|
lead.body.rstrip(),
|
|
"",
|
|
]
|
|
if lead.kind == "conjecture":
|
|
parts.extend(
|
|
[
|
|
"## Status",
|
|
"",
|
|
"**UNVERIFIED CONJECTURE — not grounded in the corpus.**",
|
|
"Promotion to the wiki requires the validation spike below to succeed.",
|
|
"",
|
|
"## Suggested validation",
|
|
"",
|
|
lead.suggested_validation,
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(parts)
|
|
|
|
|
|
def write_leads(leads: list[Lead], leads_dir: str) -> None:
|
|
"""Write ``leads`` to disk under ``leads_dir``.
|
|
|
|
* Grounded leads → ``leads_dir/grounded/<id>.md``
|
|
* Conjectures → ``leads_dir/conjectures/<id>.md`` (HARD invariant)
|
|
|
|
The function ALSO maintains an append-only ``leads_dir/INDEX.md``:
|
|
a previously-recorded id is kept on subsequent runs even if it is
|
|
overwritten with a newer body. The two sections (grounded /
|
|
conjectures) are visibly separated.
|
|
|
|
Notes
|
|
-----
|
|
* The function asserts the kind→directory routing — a conjecture
|
|
ending up under ``grounded/`` would be an invariant violation
|
|
and is raised as :class:`RuntimeError`.
|
|
"""
|
|
root = Path(leads_dir)
|
|
grounded_dir = root / "grounded"
|
|
conj_dir = root / "conjectures"
|
|
grounded_dir.mkdir(parents=True, exist_ok=True)
|
|
conj_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for lead in leads:
|
|
markdown = _lead_to_markdown(lead)
|
|
if lead.kind == "conjecture":
|
|
target_dir = conj_dir
|
|
elif lead.kind in _GROUNDED_KINDS:
|
|
target_dir = grounded_dir
|
|
else: # pragma: no cover — defensive
|
|
raise RuntimeError(f"Unknown lead kind: {lead.kind!r}")
|
|
|
|
# HARD invariant: conjectures NEVER land in grounded/ or wiki/.
|
|
if lead.kind == "conjecture" and target_dir != conj_dir:
|
|
raise RuntimeError(
|
|
f"INVARIANT VIOLATION: conjecture {lead.id} would be written "
|
|
f"to {target_dir} instead of {conj_dir}"
|
|
)
|
|
|
|
path = target_dir / f"{lead.id}.md"
|
|
path.write_text(markdown, encoding="utf-8")
|
|
|
|
# Update INDEX.md
|
|
_update_index(root)
|
|
|
|
|
|
def _update_index(root: Path) -> None:
|
|
"""Re-build ``root/INDEX.md`` from the on-disk files (full rebuild, not append-only)."""
|
|
grounded_dir = root / "grounded"
|
|
conj_dir = root / "conjectures"
|
|
grounded_files = sorted(grounded_dir.glob("*.md")) if grounded_dir.exists() else []
|
|
conj_files = sorted(conj_dir.glob("*.md")) if conj_dir.exists() else []
|
|
|
|
ts = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")
|
|
lines = [
|
|
"# Lead Index",
|
|
"",
|
|
f"_Last updated {ts} by `codex synthesis`_",
|
|
"",
|
|
"## Grounded leads (connection / gap / improvement)",
|
|
"",
|
|
]
|
|
if grounded_files:
|
|
for f in grounded_files:
|
|
title = _extract_title(f) or f.stem
|
|
lines.append(f"- `{f.stem}` — {title}")
|
|
else:
|
|
lines.append("_None._")
|
|
lines.extend(["", "## Conjectures (UNVERIFIED — quarantined here, never in wiki/)", ""])
|
|
if conj_files:
|
|
for f in conj_files:
|
|
title = _extract_title(f) or f.stem
|
|
lines.append(f"- `{f.stem}` — {title}")
|
|
else:
|
|
lines.append("_None._")
|
|
lines.append("")
|
|
(root / "INDEX.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
def _extract_title(path: Path) -> str | None:
|
|
"""Pull the H1 title from a lead markdown file, if present."""
|
|
try:
|
|
first = path.read_text(encoding="utf-8").splitlines()[0]
|
|
except (OSError, IndexError):
|
|
return None
|
|
if first.startswith("# "):
|
|
return first[2:].strip()
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Default LLM client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def default_llm_client() -> LLMClient:
|
|
"""Return an :class:`OllamaClient` configured from the synthesis settings.
|
|
|
|
Falls back to ``ollama_base_url`` when ``synthesis_llm_url`` is unset.
|
|
"""
|
|
settings = get_settings()
|
|
url = settings.synthesis_llm_url or settings.ollama_base_url
|
|
return OllamaClient(url)
|