Make source preparation preserve exact document text

docs/kb-rebuild-and-recompile.md

@@ -145,9 +145,11 @@ teleo-kb prepare-source \
 ```
 
 It queries canonical claims before extraction, caps a single document at three
-new candidates with confidence at or below `0.75`, requires exact source
+new candidates with confidence at or below `0.75`, has the model select
-substrings, records duplicate judgments, and validates a v2 manifest through
+numbered source lines, resolves those IDs to exact source substrings, records
-the source compiler. It writes private files only. A separate
+duplicate judgments, and validates a v2 manifest through the source compiler.
+Unknown line IDs are rejected rather than fuzzily matched. It writes private
+files only. A separate
 `teleo-kb propose-source` call is required to create a `pending_review` row;
 neither command applies canonical rows.
 

hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md

@@ -68,8 +68,10 @@ Apply these gates before drafting the answer or proposing a next action:
    or `source_ref` is not source identity. `teleo-kb prepare-source` is shipped
    on the VPS for a text-like artifact under the private
    `state/kb-source-inbox` root or supplied strict UTF-8
-   extraction. It retrieves canonical candidates and emits an exact-quote
+   extraction. It retrieves canonical candidates, asks the model to select
-   private manifest without a DB write. `teleo-kb propose-source` can stage one
+   numbered source lines, resolves those IDs to byte-exact text, and rejects
+   unknown IDs instead of fuzzy-matching. It emits an exact-quote private
+   manifest without a DB write. `teleo-kb propose-source` can stage one
    deterministic `pending_review` packet without apply authorization. Automatic
    Telegram attachment capture and arbitrary tweet fetching are not shipped.
    Current `approve_claim` cannot insert

hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md

@@ -61,8 +61,10 @@ Apply these before drafting the answer or proposing a next action:
    source identity. `teleo-kb prepare-source` is shipped on the VPS for a
    text-like artifact under the private `state/kb-source-inbox` root or a
    supplied strict UTF-8 extraction in that root. It reads
-   nearby canonical claims before model extraction, rejects invented quotes, and
+   nearby canonical claims before model extraction. The model selects numbered
-   emits a private manifest without a DB write. `teleo-kb propose-source` can
+   source lines; the compiler resolves those IDs back to byte-exact text and
+   rejects unknown IDs rather than fuzzy-matching invented prose. It emits a
+   private manifest without a DB write. `teleo-kb propose-source` can
    then stage one deterministic `pending_review` packet. Automatic Telegram
    attachment capture and arbitrary tweet fetching are not shipped. Approval
    begins before guarded canonical apply.
@@ -145,11 +147,19 @@ provide a separate strict UTF-8 extraction with `--text`:
 
 This performs a canonical claim retrieval before model extraction, allows at
 most three single-source candidates, caps confidence at `0.75`, binds every
-claim/evidence/duplicate judgment to an exact source substring, and writes
+claim/evidence/duplicate judgment to a numbered source line that is resolved
-private mode-`0600` outputs. It performs no database write. If status is
+deterministically to an exact substring, and writes private mode-`0600`
+outputs. It performs no database write. If status is
 `no_novel_claims`, inspect the duplicate judgments and do not stage an empty
 packet.
 
+For each newly supported document format, run one representative live
+preparation before calling it working. Verify the receipt reports both database
+and canonical apply as false, every emitted quote is an exact substring of the
+retained extraction, and canonical counts remain unchanged. After staging,
+verify only one `pending_review` proposal was added and an identical rerun is
+idempotent.
+
 After reviewing a `prepared` receipt, stage exactly one reviewable proposal:
 
 ```bash

scripts/prepare_kb_source_manifest.py

@@ -28,7 +28,8 @@ sys.path.insert(0, str(HERE))
 import compile_kb_source_packet as compiler  # noqa: E402
 
 RECEIPT_SCHEMA = "livingip.kbSourcePreparationReceipt.v1"
-MODEL_OUTPUT_SCHEMA = "livingip.kbSourceModelExtraction.v1"
+MODEL_OUTPUT_SCHEMA = "livingip.kbSourceModelExtraction.v2"
+RESOLVED_OUTPUT_SCHEMA = "livingip.kbSourceResolvedExtraction.v1"
 DEFAULT_MODEL = "anthropic/claude-sonnet-4.5"
 DEFAULT_OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
 DEFAULT_KEY_FILE = Path("/opt/teleo-eval/secrets/openrouter-key")
@@ -38,6 +39,10 @@ MAX_CANDIDATES = 20
 MAX_CLAIMS = 3
 MAX_ATTEMPTS = 2
 
+MODEL_CLAIM_KEYS = {"claim_key", "type", "text", "quote_ref", "confidence", "tags", "evidence"}
+MODEL_EVIDENCE_KEYS = {"quote_ref", "role"}
+MODEL_DUPLICATE_KEYS = {"claim_id", "source_quote_ref", "reason"}
+
 
 class PreparationError(RuntimeError):
     """Raised when a source cannot safely become a reviewable manifest."""
@@ -118,8 +123,23 @@ def _load_context(path: Path) -> dict[str, Any]:
     return {"database_search_query": query.strip(), "candidate_claims": normalized}
 
 
-def build_prompt(text: str, context: dict[str, Any], repair_error: str | None = None) -> str:
+def build_source_fragments(text: str) -> dict[str, str]:
+    """Label non-empty source lines so the model selects text instead of copying it."""
+    fragments = {
+        f"L{line_number:05d}": line
+        for line_number, line in enumerate(text.splitlines(), start=1)
+        if line.strip()
+    }
+    if not fragments:
+        raise PreparationError("extracted text has no selectable source lines")
+    return fragments
+
+
+def build_prompt(
+    fragments: dict[str, str], context: dict[str, Any], repair_error: str | None = None
+) -> str:
     candidates = json.dumps(context["candidate_claims"], indent=2, ensure_ascii=True)
+    source = "\n".join(f"[{reference}] {line}" for reference, line in fragments.items())
     repair = ""
     if repair_error:
         repair = (
@@ -136,7 +156,8 @@ Canonical claims retrieved before extraction:
 Rules:
 - Return 0 to {MAX_CLAIMS} genuinely novel, arguable claims. A fact, summary, heading, implementation detail, or restatement is not automatically a claim.
 - Prefer no new claim when a canonical candidate already makes the same argument. Record that in duplicates instead.
-- Every claim quote, evidence quote, and duplicate source_quote must be copied byte-for-byte from SOURCE and be a contiguous exact substring.
+- Select quote_ref and source_quote_ref only from the bracketed line IDs in SOURCE. Do not copy or rewrite source prose.
+- Each selected line becomes the exact quote deterministically after your response.
 - A single source cannot justify confidence above 0.75.
 - claim type must be one of: {", ".join(sorted(compiler.ap.CLAIM_TYPES))}.
 - evidence role must be one of: {", ".join(sorted(compiler.ap.EVIDENCE_ROLES))}.
@@ -152,20 +173,20 @@ Return exactly this JSON object and no prose:
       "claim_key": "stable_key",
       "type": "empirical|structural|normative|meta|concept",
       "text": "concise proposition",
-      "quote": "exact source substring",
+      "quote_ref": "L00001",
       "confidence": 0.0,
       "tags": ["tag"],
-      "evidence": [{{"quote": "exact source substring", "role": "grounds|illustrates|contradicts"}}]
+      "evidence": [{{"quote_ref": "L00001", "role": "grounds|illustrates|contradicts"}}]
     }}
   ],
   "duplicates": [
-    {{"claim_id": "retrieved canonical UUID", "source_quote": "exact source substring", "reason": "same argument"}}
+    {{"claim_id": "retrieved canonical UUID", "source_quote_ref": "L00001", "reason": "same argument"}}
   ],
   "extraction_notes": "short reviewer-facing note"
 }}
 {repair}
 SOURCE START
-{text}
+{source}
 SOURCE END
 """
 
@@ -251,14 +272,96 @@ def parse_model_output(content: str) -> dict[str, Any]:
     for index, claim in enumerate(claims):
         if not isinstance(claim, dict):
             raise PreparationError(f"model extraction claims[{index}] must be an object")
+        if set(claim) != MODEL_CLAIM_KEYS:
+            raise PreparationError(
+                f"model extraction claims[{index}] keys must equal {sorted(MODEL_CLAIM_KEYS)}"
+            )
         confidence = claim.get("confidence")
         if confidence is not None and (
             isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 0.75
         ):
             raise PreparationError(f"model extraction claims[{index}].confidence must be between 0 and 0.75 or null")
+        evidence = claim.get("evidence")
+        if not isinstance(evidence, list) or not evidence:
+            raise PreparationError(f"model extraction claims[{index}].evidence must be a non-empty list")
+        for evidence_index, row in enumerate(evidence):
+            if not isinstance(row, dict) or set(row) != MODEL_EVIDENCE_KEYS:
+                raise PreparationError(
+                    f"model extraction claims[{index}].evidence[{evidence_index}] keys must equal "
+                    f"{sorted(MODEL_EVIDENCE_KEYS)}"
+                )
+    for index, duplicate in enumerate(duplicates):
+        if not isinstance(duplicate, dict) or set(duplicate) != MODEL_DUPLICATE_KEYS:
+            raise PreparationError(
+                f"model extraction duplicates[{index}] keys must equal {sorted(MODEL_DUPLICATE_KEYS)}"
+            )
     return value
 
 
+def _resolve_reference(reference: Any, fragments: dict[str, str], label: str) -> str:
+    if not isinstance(reference, str) or reference not in fragments:
+        raise PreparationError(f"{label} must select one of the bracketed SOURCE line IDs")
+    return fragments[reference]
+
+
+def resolve_model_output(selection: dict[str, Any], fragments: dict[str, str]) -> dict[str, Any]:
+    claims: list[dict[str, Any]] = []
+    selected_references: list[dict[str, str]] = []
+    for claim_index, claim in enumerate(selection["claims"]):
+        quote_ref = claim["quote_ref"]
+        quote = _resolve_reference(quote_ref, fragments, f"claims[{claim_index}].quote_ref")
+        selected_references.append({"kind": "claim", "reference": quote_ref})
+        evidence: list[dict[str, str]] = []
+        for evidence_index, row in enumerate(claim["evidence"]):
+            evidence_ref = row["quote_ref"]
+            evidence.append(
+                {
+                    "quote": _resolve_reference(
+                        evidence_ref,
+                        fragments,
+                        f"claims[{claim_index}].evidence[{evidence_index}].quote_ref",
+                    ),
+                    "role": row["role"],
+                }
+            )
+            selected_references.append({"kind": "evidence", "reference": evidence_ref})
+        claims.append(
+            {
+                "claim_key": claim["claim_key"],
+                "type": claim["type"],
+                "text": claim["text"],
+                "quote": quote,
+                "confidence": claim["confidence"],
+                "tags": claim["tags"],
+                "evidence": evidence,
+            }
+        )
+
+    duplicates: list[dict[str, str]] = []
+    for duplicate_index, duplicate in enumerate(selection["duplicates"]):
+        source_quote_ref = duplicate["source_quote_ref"]
+        duplicates.append(
+            {
+                "claim_id": duplicate["claim_id"],
+                "source_quote": _resolve_reference(
+                    source_quote_ref,
+                    fragments,
+                    f"duplicates[{duplicate_index}].source_quote_ref",
+                ),
+                "reason": duplicate["reason"],
+            }
+        )
+        selected_references.append({"kind": "duplicate", "reference": source_quote_ref})
+
+    return {
+        "schema": RESOLVED_OUTPUT_SCHEMA,
+        "claims": claims,
+        "duplicates": duplicates,
+        "extraction_notes": selection["extraction_notes"],
+        "selected_source_references": selected_references,
+    }
+
+
 def build_manifest(
     *,
     args: argparse.Namespace,
@@ -324,6 +427,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
     artifact_bytes = _read_nonempty(args.artifact, "artifact")
     text_bytes = extract_utf8_text(args.artifact, args.text)
     text = text_bytes.decode("utf-8", errors="strict")
+    fragments = build_source_fragments(text)
     context = _load_context(args.kb_context)
 
     requested_output_dir = args.output_dir.expanduser()
@@ -351,7 +455,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
     manifest: dict[str, Any] | None = None
     for attempt in range(1, MAX_ATTEMPTS + 1):
         raw, usage = call_openrouter(
-            build_prompt(text, context, repair_error),
+            build_prompt(fragments, context, repair_error),
             model=args.model,
             key_file=args.key_file,
             url=args.openrouter_url,
@@ -359,7 +463,8 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
         )
         attempts.append({"attempt": attempt, "response_sha256": sha256_bytes(raw.encode("utf-8")), "usage": usage})
         try:
-            extraction = parse_model_output(raw)
+            selection = parse_model_output(raw)
+            extraction = resolve_model_output(selection, fragments)
             compiler._validate_dedupe(
                 {
                     "database_search_query": context["database_search_query"],
@@ -440,6 +545,8 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
             "model": args.model,
             "attempts": attempts,
             "claim_count": len(extraction["claims"]),
+            "selectable_source_line_count": len(fragments),
+            "selected_source_references": extraction["selected_source_references"],
             "notes": extraction["extraction_notes"],
         },
         "outputs": {

tests/test_prepare_kb_source_manifest.py

@@ -66,19 +66,19 @@ def _inputs(tmp_path: Path) -> tuple[Path, Path]:
     return artifact, context
 
 
-def _model_output(*, claims: bool = True, bad_quote: bool = False) -> str:
+def _model_output(*, claims: bool = True, bad_reference: bool = False) -> str:
     claim_rows = []
     if claims:
-        quote = "Invented quote." if bad_quote else CLAIM_QUOTE
+        quote_ref = "L99999" if bad_reference else "L00005"
         claim_rows = [
             {
                 "claim_key": "exact_quotes_bind_prepared_claims",
                 "type": "structural",
                 "text": "Prepared claims remain auditable when their evidence is bound to exact source quotes.",
-                "quote": quote,
+                "quote_ref": quote_ref,
                 "confidence": 0.7,
                 "tags": ["provenance", "review"],
-                "evidence": [{"quote": quote, "role": "grounds"}],
+                "evidence": [{"quote_ref": quote_ref, "role": "grounds"}],
             }
         ]
     return json.dumps(
@@ -88,7 +88,7 @@ def _model_output(*, claims: bool = True, bad_quote: bool = False) -> str:
             "duplicates": [
                 {
                     "claim_id": CANDIDATE_ID,
-                    "source_quote": DUPLICATE_QUOTE,
+                    "source_quote_ref": "L00003",
                     "reason": "The canonical candidate already captures the review-before-apply argument.",
                 }
             ],
@@ -120,14 +120,23 @@ def test_prepare_builds_private_compiler_validated_manifest_without_db_write(
     manifest = json.loads(Path(receipt["outputs"]["manifest"]).read_text(encoding="utf-8"))
     assert manifest["schema"] == preparer.compiler.MANIFEST_SCHEMA_V2
     assert manifest["dedupe"]["duplicates"][0]["claim_id"] == CANDIDATE_ID
+    assert manifest["claims"][0]["quote"] == CLAIM_QUOTE
+    assert manifest["dedupe"]["duplicates"][0]["source_quote"] == DUPLICATE_QUOTE
+    assert receipt["extraction"]["selected_source_references"] == [
+        {"kind": "claim", "reference": "L00005"},
+        {"kind": "evidence", "reference": "L00005"},
+        {"kind": "duplicate", "reference": "L00003"},
+    ]
     assert stat.S_IMODE(output_dir.stat().st_mode) == 0o700
     for path in output_dir.iterdir():
         assert stat.S_IMODE(path.stat().st_mode) == 0o600
 
 
-def test_prepare_retries_once_when_model_invents_a_quote(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_prepare_retries_once_when_model_invents_a_source_reference(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
     artifact, context = _inputs(tmp_path)
-    outputs = iter((_model_output(bad_quote=True), _model_output()))
+    outputs = iter((_model_output(bad_reference=True), _model_output()))
     prompts: list[str] = []
 
     def fake_call(prompt: str, **_kwargs):
@@ -141,7 +150,20 @@ def test_prepare_retries_once_when_model_invents_a_quote(tmp_path: Path, monkeyp
     assert receipt["status"] == "prepared"
     assert len(receipt["extraction"]["attempts"]) == 2
     assert "previous extraction failed deterministic validation" in prompts[1]
-    assert "not an exact substring" in prompts[1]
+    assert "must select one of the bracketed SOURCE line IDs" in prompts[1]
+
+
+def test_source_fragments_preserve_exact_unicode_and_punctuation() -> None:
+    text = "Heading\n\nIdentity pins use an exact version - never a floating alias.\nCurly: \u201creviewed\u201d.\n"
+
+    fragments = preparer.build_source_fragments(text)
+
+    assert fragments == {
+        "L00001": "Heading",
+        "L00003": "Identity pins use an exact version - never a floating alias.",
+        "L00004": "Curly: \u201creviewed\u201d.",
+    }
+    assert all(value in text for value in fragments.values())
 
 
 def test_prepare_returns_no_novel_claims_without_manifest_or_stageable_packet(