diff --git a/services/proposal-engine/.gitignore b/services/proposal-engine/.gitignore new file mode 100644 index 00000000..4c3f8485 --- /dev/null +++ b/services/proposal-engine/.gitignore @@ -0,0 +1,8 @@ +.venv/ +__pycache__/ +*.pyc +.mypy_cache/ +.pytest_cache/ +.ruff_cache/ +.coverage +*.egg-info/ diff --git a/services/proposal-engine/GROUNDING.md b/services/proposal-engine/GROUNDING.md new file mode 100644 index 00000000..08264485 --- /dev/null +++ b/services/proposal-engine/GROUNDING.md @@ -0,0 +1,66 @@ +# The grounding contract — `proposal_engine` + +The winniio-proposal golden rule — *"every client-specific claim traces to a +verbatim source; if it is not traceable, it does not go in"* — is now a **typed +contract**, not a prose discipline. + +`parse_appendix(rows, corpus)` grounds every appendix row against the client's +own words: + +| Error | Means | +|---|---| +| `UngroundedClaim` | the row's quote is **not verbatim-present** in the source it names — fabrication is a parse error | +| `MalformedRow` | missing field, wrong type, unknown source id, or a SMILE stage outside the canonical six | + +A `GroundedProposal`, if it exists at all, has every row proven grounded. +`to_markdown()` renders the canonical *"what we heard → where it lands → SMILE +stage → source"* appendix table unchanged, so the skill's output is identical. + +This is the **NOOA lesson** (typed returns / PredictStrategy — *no model-written +code executed*) that landed in MENTOR's judge (`Life-Atlas/mentor` #22), applied +to the proposal engine: the same shape that made a fabricated axiom a type error +makes a fabricated client claim one. + +## Honest boundary (zero-gaslight) + +The contract proves a quote is **present**, not that it is **faithfully +represented**. A real fragment recombined to mislead — dropping a negation, +splicing across a clause — still grounds. This limitation is pinned in +`tests/test_boundary.py`. The human roast + final consistency sweep in the skill +remain the backstop for *misquotation*. What the contract kills cold is *pure +invention* (a claim with no verbatim basis at all — the dominant LLM failure). + +**Matching is deliberately forgiving on form, strict on substance.** Grounding +compares after folding typographic punctuation (curly quotes, em/en dashes, +ellipsis, nbsp), collapsing whitespace, and casefolding — so a real quote an LLM +retyped with straight ASCII punctuation or a capitalized sentence-start is not +falsely rejected. It does **not** loosen what must exist: invented text appears +in no source under any normalization. + +**Still unmeasured (honest gap):** the real false-reject rate against actual +Fireflies transcripts + a real appendix has not been measured — only synthetic +sources are unit-tested. That measurement is the fast-follow before this gate is +wired into the headless flow. + +## Run + +```bash +pip install -e ".[dev]" +PYTHONPATH=src pytest # 26 tests: contract + boundary + api +``` + +HTTP gate (roadmap tier 3 — the headless door): + +```bash +pip install -e ".[api]" +uvicorn proposal_engine.api:app +# POST /appendix/validate {"sources":[{id,kind,text}], "appendix":[{heard,source_id,lands,smile_stage}]} +# 200 -> {grounded:true, markdown} 422 -> {grounded:false, error_type, detail} +``` + +## Deliberately not built yet (KISS) + +No auth, no persistence, no GraphQL, no A2A, no vector DB. The contract is the +foundation; those arrive with real tenants and a real headless flow, not before. +The generation loop itself stays in the `winniio-proposal` skill — this package +is the **gate that loop must pass** before a proposal ships. diff --git a/services/proposal-engine/pyproject.toml b/services/proposal-engine/pyproject.toml new file mode 100644 index 00000000..87be8839 --- /dev/null +++ b/services/proposal-engine/pyproject.toml @@ -0,0 +1,37 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "proposal-engine" +version = "0.1.0" +description = "North Star proposal engine — the grounding contract. Every client claim traces to a verbatim source, or it is a type error." +requires-python = ">=3.11" +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "ruff>=0.4.0", + "mypy>=1.10.0", +] +api = [ + "fastapi>=0.110", +] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP"] +ignore = ["E501"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.mypy] +ignore_missing_imports = true diff --git a/services/proposal-engine/src/proposal_engine/__init__.py b/services/proposal-engine/src/proposal_engine/__init__.py new file mode 100644 index 00000000..5ec0ae44 --- /dev/null +++ b/services/proposal-engine/src/proposal_engine/__init__.py @@ -0,0 +1,30 @@ +"""North Star proposal engine — the grounding contract. + +Public surface: the typed appendix contract that makes an ungrounded proposal +claim a parse error rather than a discipline failure. +""" + +from proposal_engine.appendix import ( + AppendixError, + AppendixRow, + GroundedProposal, + MalformedRow, + UngroundedClaim, + parse_appendix, +) +from proposal_engine.corpus import Source, SourceCorpus +from proposal_engine.smile import SMILE_STAGES, SmileStage, is_valid_stage + +__all__ = [ + "AppendixError", + "AppendixRow", + "GroundedProposal", + "MalformedRow", + "UngroundedClaim", + "parse_appendix", + "Source", + "SourceCorpus", + "SMILE_STAGES", + "SmileStage", + "is_valid_stage", +] diff --git a/services/proposal-engine/src/proposal_engine/api.py b/services/proposal-engine/src/proposal_engine/api.py new file mode 100644 index 00000000..647c23f4 --- /dev/null +++ b/services/proposal-engine/src/proposal_engine/api.py @@ -0,0 +1,71 @@ +"""Thin HTTP surface for the grounding contract — the headless (tier 3) door. + +One endpoint: POST /appendix/validate takes {sources, appendix} and returns +either the grounded markdown table or a structured grounding error. This is the +gate the roadmap's "questionnaire JSON → FastAPI → claude -p" flow calls before +any proposal is allowed to ship. Kept deliberately minimal (KISS): no auth, no +persistence, no GraphQL — those arrive with real tenants, not before. + +Requires the [api] extra (`pip install -e .[api]`); the core contract has no +web dependency, so tests and the skill can use it without FastAPI installed. +""" + +from __future__ import annotations + +from typing import Any + +try: + from fastapi import FastAPI + from fastapi.responses import JSONResponse +except ImportError as e: # pragma: no cover - exercised only without the extra + raise ImportError("proposal_engine.api requires the [api] extra: pip install -e '.[api]'") from e + +from proposal_engine.appendix import AppendixError, UngroundedClaim, parse_appendix +from proposal_engine.corpus import Source, SourceCorpus + +app = FastAPI(title="North Star Proposal Engine — grounding gate", version="0.1.0") + + +def _build_corpus(sources: list[dict[str, Any]]) -> SourceCorpus: + return SourceCorpus.from_sources( + [Source(id=s["id"], kind=s.get("kind", "unknown"), text=s["text"]) for s in sources] + ) + + +@app.post("/appendix/validate") +def validate_appendix(payload: dict[str, Any]) -> JSONResponse: + """Ground an appendix against provided sources. + + Body: {"sources": [{"id","kind","text"}...], "appendix": [{row}...]}. + 200 -> {"grounded": true, "markdown": ...} + 422 -> {"grounded": false, "error_type": ..., "detail": ...} + """ + try: + corpus = _build_corpus(payload["sources"]) + except (KeyError, TypeError, ValueError) as e: + return JSONResponse( + status_code=400, content={"grounded": False, "error_type": "bad_request", "detail": str(e)} + ) + + try: + grounded = parse_appendix(payload["appendix"], corpus) + except UngroundedClaim as e: + return JSONResponse( + status_code=422, + content={"grounded": False, "error_type": "ungrounded_claim", "detail": str(e)}, + ) + except AppendixError as e: + return JSONResponse( + status_code=422, + content={"grounded": False, "error_type": "malformed", "detail": str(e)}, + ) + except KeyError as e: + return JSONResponse( + status_code=400, + content={"grounded": False, "error_type": "bad_request", "detail": f"missing {e}"}, + ) + + return JSONResponse( + status_code=200, + content={"grounded": True, "rows": len(grounded.rows), "markdown": grounded.to_markdown()}, + ) diff --git a/services/proposal-engine/src/proposal_engine/appendix.py b/services/proposal-engine/src/proposal_engine/appendix.py new file mode 100644 index 00000000..44587445 --- /dev/null +++ b/services/proposal-engine/src/proposal_engine/appendix.py @@ -0,0 +1,151 @@ +"""The appendix contract — grounding as a type, not a discipline. + +The winniio-proposal appendix is the proof a proposal is grounded: one row per +client point, "what we heard → where it lands → SMILE stage", each traced to a +verbatim source. Prose discipline used to enforce it. Here it is a typed +contract: an appendix row whose quote is not verbatim-present in the named +source raises `UngroundedClaim`; a bad SMILE stage or missing field raises +`MalformedRow`. Fabrication becomes a parse error — the exact NOOA lesson +(PredictStrategy, typed returns, no model-written code executed) that landed in +MENTOR's judge, applied to the proposal engine. + +`GroundedProposal.to_markdown()` renders the canonical appendix table, so the +skill's output shape is unchanged. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass + +from proposal_engine.corpus import SourceCorpus +from proposal_engine.smile import SMILE_STAGES, SmileStage + +_JSON_BLOCK_HINT = "expected a JSON array of appendix rows" + + +class AppendixError(ValueError): + """Base: the appendix is not a valid, fully-grounded set of rows.""" + + +class MalformedRow(AppendixError): + """A row is missing a field, has a wrong type, or a bad SMILE stage.""" + + +class UngroundedClaim(AppendixError): + """A row's quote is not verbatim-present in the source it names — the exact + failure the golden rule exists to catch, turned into a type error.""" + + +@dataclass(frozen=True) +class AppendixRow: + heard: str # the client's verbatim words + source_id: str # the source those words came from + lands: str # where in the proposal this point is used + smile_stage: SmileStage + + +@dataclass(frozen=True) +class GroundedProposal: + """An appendix proven grounded against a corpus. If it exists, every row's + `heard` was found verbatim in its named source.""" + + rows: list[AppendixRow] + + def to_markdown(self) -> str: + lines = [ + "| What we heard | Where it lands | SMILE stage | Source |", + "|---|---|---|---|", + ] + for r in self.rows: + heard = r.heard.replace("|", "\\|").replace("\n", " ").strip() + lands = r.lands.replace("|", "\\|").replace("\n", " ").strip() + lines.append(f"| {heard} | {lands} | {r.smile_stage} | {r.source_id} |") + return "\n".join(lines) + + +def _require(cond: bool, msg: str, exc: type[AppendixError] = MalformedRow) -> None: + if not cond: + raise exc(msg) + + +def _coerce_rows(raw: str | list) -> list: + if isinstance(raw, str): + try: + parsed = json.loads(raw) + except json.JSONDecodeError as e: + raise MalformedRow(f"{_JSON_BLOCK_HINT}: {e}") from e + else: + parsed = raw + _require(isinstance(parsed, list), _JSON_BLOCK_HINT) + return parsed + + +def _parse_row(item: object, corpus: SourceCorpus, pos: int) -> AppendixRow: + _require(isinstance(item, dict), f"row {pos} is not an object") + assert isinstance(item, dict) + for key in ("heard", "source_id", "lands", "smile_stage"): + _require(key in item, f"row {pos} missing '{key}'") + + heard = item["heard"] + source_id = item["source_id"] + lands = item["lands"] + stage = item["smile_stage"] + + _require( + isinstance(heard, str) and heard.strip() != "", + f"row {pos} 'heard' must be a non-empty string", + ) + _require( + isinstance(source_id, str) and source_id.strip() != "", + f"row {pos} 'source_id' must be a non-empty string", + ) + _require( + isinstance(lands, str) and lands.strip() != "", + f"row {pos} 'lands' must be a non-empty string", + ) + _require( + isinstance(stage, str) and stage in SMILE_STAGES, + f"row {pos} 'smile_stage' must be one of {list(SMILE_STAGES)}", + ) + + # Grounding, in order of specificity so the error message is the most useful: + _require( + corpus.has(source_id), + f"row {pos} names source {source_id!r}, which is not in the corpus", + ) + _require( + SourceCorpus.is_substantive(heard), + f"row {pos} quote is too short/generic to ground a claim: {heard!r}", + UngroundedClaim, + ) + if not corpus.contains(heard, source_id): + elsewhere = corpus.contains_anywhere(heard) + hint = ( + f" (it does appear in {elsewhere!r} — name that source)" + if elsewhere + else " (not found in any source — did the client actually say this?)" + ) + raise UngroundedClaim( + f"row {pos} quote is not verbatim in source {source_id!r}{hint}: {heard!r}" + ) + + return AppendixRow( + heard=heard.strip(), + source_id=source_id.strip(), + lands=lands.strip(), + smile_stage=stage, # type: ignore[arg-type] # validated against Literal above + ) + + +def parse_appendix(raw: str | list, corpus: SourceCorpus) -> GroundedProposal: + """Parse + ground an appendix against the source corpus. + + Raises MalformedRow (shape/stage) or UngroundedClaim (quote not verbatim in + its named source). A caller generating the appendix from an LLM turns either + into a retry-as-observation, exactly as MENTOR's judge loop does. + """ + rows_raw = _coerce_rows(raw) + _require(len(rows_raw) >= 1, "appendix must have at least one row") + rows = [_parse_row(item, corpus, i) for i, item in enumerate(rows_raw, start=1)] + return GroundedProposal(rows=rows) diff --git a/services/proposal-engine/src/proposal_engine/corpus.py b/services/proposal-engine/src/proposal_engine/corpus.py new file mode 100644 index 00000000..a7548cb0 --- /dev/null +++ b/services/proposal-engine/src/proposal_engine/corpus.py @@ -0,0 +1,105 @@ +"""The source corpus — the client's verbatim words, and only those. + +The winniio-proposal golden rule: "Every client-specific claim traces to a +verbatim source. If it is not traceable, it does not go in." This module is the +mechanical check behind that rule — a claim is grounded only if its quote +appears, verbatim (whitespace-normalized), inside a named source the client +actually produced (transcript, email, RFP). + +Whitespace is normalized (transcripts wrap and re-indent); case is preserved +(names, product terms, and the client's exact framing are load-bearing). A +too-short quote is rejected as non-grounding: matching "the" proves nothing. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +# A quote must carry enough signal to be evidence. Tuned low enough to allow a +# short but specific client phrase ("fire us and keep everything"), high enough +# that a stopword or two cannot ground a fabricated claim. +_MIN_QUOTE_CHARS = 12 +_MIN_QUOTE_WORDS = 3 + +_WS = re.compile(r"\s+") + +# Fold typographic variants an LLM introduces when it retypes a client quote, +# so a real, grounded quote is not falsely rejected over a curly apostrophe or +# an em-dash the transcript happened to render differently. Pure invention is +# unaffected — invented text appears in no source regardless of punctuation. +_PUNCT_FOLD = str.maketrans( + { + "‘": "'", "’": "'", "‛": "'", # ‘ ’ ‛ -> ' + "“": '"', "”": '"', "„": '"', # “ ” „ -> " + "–": "-", "—": "-", "―": "-", # – — ― -> - + "…": "...", # … -> ... + " ": " ", # nbsp -> space + } +) + + +def _normalize(text: str) -> str: + """Canonicalize for grounding comparison: fold typographic punctuation, + collapse whitespace, casefold. Widens what counts as a verbatim match to + survive LLM re-typing; it does NOT let invented claims through.""" + return _WS.sub(" ", text.translate(_PUNCT_FOLD)).strip().casefold() + + +@dataclass(frozen=True) +class Source: + """One verbatim client artifact — a transcript, an email, an RFP.""" + + id: str + kind: str # "transcript" | "email" | "rfp" | ... + text: str + + @property + def normalized(self) -> str: + return _normalize(self.text) + + +@dataclass +class SourceCorpus: + """All verbatim sources for one opportunity, keyed by id.""" + + sources: dict[str, Source] = field(default_factory=dict) + + @classmethod + def from_sources(cls, sources: list[Source]) -> SourceCorpus: + by_id: dict[str, Source] = {} + for s in sources: + if s.id in by_id: + raise ValueError(f"duplicate source id: {s.id!r}") + by_id[s.id] = s + return cls(sources=by_id) + + def has(self, source_id: str) -> bool: + return source_id in self.sources + + @staticmethod + def is_substantive(quote: str) -> bool: + """A quote long/specific enough to count as grounding evidence.""" + norm = _normalize(quote) + return len(norm) >= _MIN_QUOTE_CHARS and len(norm.split(" ")) >= _MIN_QUOTE_WORDS + + def contains(self, quote: str, source_id: str) -> bool: + """True iff `quote` appears verbatim (whitespace-normalized) in the + named source AND is substantive. A short or absent quote is not + grounding.""" + src = self.sources.get(source_id) + if src is None or not self.is_substantive(quote): + return False + return _normalize(quote) in src.normalized + + def contains_anywhere(self, quote: str) -> str | None: + """Return the id of the first source containing the quote, else None. + Used for diagnostics — grounding still requires the row to name the + correct source explicitly.""" + if not self.is_substantive(quote): + return None + needle = _normalize(quote) + for sid, src in self.sources.items(): + if needle in src.normalized: + return sid + return None diff --git a/services/proposal-engine/src/proposal_engine/smile.py b/services/proposal-engine/src/proposal_engine/smile.py new file mode 100644 index 00000000..2caad9bb --- /dev/null +++ b/services/proposal-engine/src/proposal_engine/smile.py @@ -0,0 +1,25 @@ +"""The SMILE journey — the six stages every appendix row must map to. + +SMILE (Waern 2026, DOI 10.5281/zenodo.20175406) is the spine of every WINNIIO +proposal. An appendix row that claims a stage outside these six is malformed — +a typed error, not a silent typo that ships. +""" + +from __future__ import annotations + +from typing import Literal, get_args + +SmileStage = Literal[ + "Reality Emulation", + "Concurrent Engineering", + "Collective Intelligence", + "Contextual Intelligence", + "Continuous Intelligence", + "Perpetual Wisdom", +] + +SMILE_STAGES: tuple[str, ...] = get_args(SmileStage) + + +def is_valid_stage(stage: str) -> bool: + return stage in SMILE_STAGES diff --git a/services/proposal-engine/tests/test_api.py b/services/proposal-engine/tests/test_api.py new file mode 100644 index 00000000..b4a86d1d --- /dev/null +++ b/services/proposal-engine/tests/test_api.py @@ -0,0 +1,61 @@ +"""API surface tests — skipped cleanly if the [api] extra isn't installed.""" + +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("httpx") # starlette TestClient transport + +from fastapi.testclient import TestClient # noqa: E402 + +from proposal_engine.api import app # noqa: E402 + +client = TestClient(app) + +SOURCES = [ + { + "id": "call-1", + "kind": "transcript", + "text": "Client: the driver is re-acknowledgement by the funders before the review.", + } +] + + +def test_grounded_appendix_returns_200_markdown(): + body = { + "sources": SOURCES, + "appendix": [ + { + "heard": "re-acknowledgement by the funders before the review", + "source_id": "call-1", + "lands": "Executive summary", + "smile_stage": "Reality Emulation", + } + ], + } + r = client.post("/appendix/validate", json=body) + assert r.status_code == 200 + data = r.json() + assert data["grounded"] is True + assert "SMILE stage" in data["markdown"] + + +def test_ungrounded_claim_returns_422(): + body = { + "sources": SOURCES, + "appendix": [ + { + "heard": "the client signed a five year exclusive contract", + "source_id": "call-1", + "lands": "Commercial", + "smile_stage": "Reality Emulation", + } + ], + } + r = client.post("/appendix/validate", json=body) + assert r.status_code == 422 + assert r.json()["error_type"] == "ungrounded_claim" + + +def test_missing_sources_is_400(): + r = client.post("/appendix/validate", json={"appendix": []}) + assert r.status_code == 400 diff --git a/services/proposal-engine/tests/test_appendix.py b/services/proposal-engine/tests/test_appendix.py new file mode 100644 index 00000000..073e7a88 --- /dev/null +++ b/services/proposal-engine/tests/test_appendix.py @@ -0,0 +1,133 @@ +"""Appendix contract tests — the core: a fabricated claim is a parse error.""" + +import json + +import pytest + +from proposal_engine.appendix import ( + AppendixError, + GroundedProposal, + MalformedRow, + UngroundedClaim, + parse_appendix, +) +from proposal_engine.corpus import Source, SourceCorpus + +TRANSCRIPT = ( + "Client: The real driver is we need to be re-acknowledged by the funders before " + "the autumn review. If we can't articulate the value outward, the programme is at risk." +) +EMAIL = "We would want everything hosted in Switzerland, and no vendor lock-in whatsoever." + + +def _corpus() -> SourceCorpus: + return SourceCorpus.from_sources( + [ + Source(id="call-06-02", kind="transcript", text=TRANSCRIPT), + Source(id="email-06-05", kind="email", text=EMAIL), + ] + ) + + +def _row(**over) -> dict: + base = { + "heard": "re-acknowledged by the funders before the autumn review", + "source_id": "call-06-02", + "lands": "Executive summary — the one outcome", + "smile_stage": "Reality Emulation", + } + base.update(over) + return base + + +def test_grounded_appendix_parses(): + gp = parse_appendix([_row()], _corpus()) + assert isinstance(gp, GroundedProposal) + assert len(gp.rows) == 1 + assert gp.rows[0].source_id == "call-06-02" + + +def test_accepts_json_string_input(): + gp = parse_appendix(json.dumps([_row()]), _corpus()) + assert len(gp.rows) == 1 + + +def test_fabricated_quote_is_ungrounded(): + with pytest.raises(UngroundedClaim): + parse_appendix( + [_row(heard="the client committed to a three year exclusive deal")], _corpus() + ) + + +def test_quote_in_wrong_source_names_the_right_one(): + with pytest.raises(UngroundedClaim) as exc: + parse_appendix( + [_row(heard="no vendor lock-in whatsoever", source_id="call-06-02")], _corpus() + ) + assert "email-06-05" in str(exc.value) # error points to the correct source + + +def test_quote_from_second_source_grounds_when_named_correctly(): + gp = parse_appendix( + [ + _row( + heard="everything hosted in Switzerland", + source_id="email-06-05", + smile_stage="Contextual Intelligence", + ) + ], + _corpus(), + ) + assert gp.rows[0].source_id == "email-06-05" + + +def test_unknown_source_id_is_malformed(): + with pytest.raises(MalformedRow): + parse_appendix([_row(source_id="nonexistent")], _corpus()) + + +def test_bad_smile_stage_is_malformed(): + with pytest.raises(MalformedRow): + parse_appendix([_row(smile_stage="Synergy Realization")], _corpus()) + + +def test_missing_field_is_malformed(): + r = _row() + del r["lands"] + with pytest.raises(MalformedRow): + parse_appendix([r], _corpus()) + + +def test_too_short_quote_is_ungrounded(): + with pytest.raises(UngroundedClaim): + parse_appendix([_row(heard="the funders")], _corpus()) + + +def test_empty_appendix_rejected(): + with pytest.raises(MalformedRow): + parse_appendix([], _corpus()) + + +def test_non_list_rejected(): + with pytest.raises(MalformedRow): + parse_appendix(json.dumps({"heard": "x"}), _corpus()) + + +def test_errors_share_base_for_broad_catch(): + assert issubclass(MalformedRow, AppendixError) + assert issubclass(UngroundedClaim, AppendixError) + + +def test_to_markdown_renders_canonical_table(): + gp = parse_appendix([_row()], _corpus()) + md = gp.to_markdown() + assert "What we heard | Where it lands | SMILE stage | Source" in md + assert "Reality Emulation" in md + assert "call-06-02" in md + + +def test_to_markdown_escapes_pipes(): + gp = parse_appendix( + [_row(lands="Governance | Commercial")], _corpus() + ) + assert "Governance \\| Commercial" in gp.to_markdown() diff --git a/services/proposal-engine/tests/test_boundary.py b/services/proposal-engine/tests/test_boundary.py new file mode 100644 index 00000000..838076e3 --- /dev/null +++ b/services/proposal-engine/tests/test_boundary.py @@ -0,0 +1,64 @@ +"""Honest boundary tests — what the grounding contract does NOT catch. + +Zero-gaslight: the contract proves a quote is verbatim-present in a named +source. It does NOT prove the quote preserves the source's meaning. A real +fragment can be recombined to mislead (dropping a negation, splicing across a +clause). These tests PIN that limitation so no one mistakes "grounded" for +"honestly represented" — the human roast + consistency sweep in the skill +remains the backstop for misquotation. +""" + +from proposal_engine.appendix import parse_appendix +from proposal_engine.corpus import Source, SourceCorpus + + +def test_negation_stripping_is_NOT_caught_known_limitation(): + """Source says the client does NOT want something; a row quoting only the + positive fragment still grounds. Documented gap, not a bug in presence.""" + corpus = SourceCorpus.from_sources( + [ + Source( + id="call-1", + kind="transcript", + text="Client: We would not want everything hosted in the public cloud.", + ) + ] + ) + # The misleading fragment IS verbatim-present, so it grounds. This is the + # known limitation: presence != faithful representation. + gp = parse_appendix( + [ + { + "heard": "want everything hosted in the public cloud", + "source_id": "call-1", + "lands": "Governance", + "smile_stage": "Contextual Intelligence", + } + ], + corpus, + ) + assert len(gp.rows) == 1 # grounds — hence the human-review backstop exists + + +def test_pure_invention_IS_caught(): + """The 90% case the contract exists for: a wholly invented claim with no + verbatim basis is rejected.""" + import pytest + + from proposal_engine.appendix import UngroundedClaim + + corpus = SourceCorpus.from_sources( + [Source(id="call-1", kind="transcript", text="Client: we care about data sovereignty.")] + ) + with pytest.raises(UngroundedClaim): + parse_appendix( + [ + { + "heard": "the client agreed to a 250,000 euro annual retainer", + "source_id": "call-1", + "lands": "Commercial", + "smile_stage": "Reality Emulation", + } + ], + corpus, + ) diff --git a/services/proposal-engine/tests/test_corpus.py b/services/proposal-engine/tests/test_corpus.py new file mode 100644 index 00000000..724ef5bb --- /dev/null +++ b/services/proposal-engine/tests/test_corpus.py @@ -0,0 +1,78 @@ +"""Corpus grounding tests — verbatim presence, whitespace tolerance, floors.""" + +import pytest + +from proposal_engine.corpus import Source, SourceCorpus + +TRANSCRIPT = ( + "Nicolas: So the real driver is we need to be re-acknowledged by the funders " + "before the autumn review.\n" + "Client: Exactly. And honestly, if we can't articulate the value outward, the " + "whole programme is at risk." +) + + +def _corpus() -> SourceCorpus: + return SourceCorpus.from_sources( + [Source(id="call-2026-06-02", kind="transcript", text=TRANSCRIPT)] + ) + + +def test_verbatim_quote_is_grounded(): + c = _corpus() + assert c.contains("re-acknowledged by the funders before the autumn review", "call-2026-06-02") + + +def test_whitespace_differences_are_tolerated(): + c = _corpus() + # newline/indent collapsed — transcripts wrap arbitrarily + assert c.contains("articulate the value outward", "call-2026-06-02") + + +def test_absent_quote_is_not_grounded(): + c = _corpus() + assert not c.contains("we have a signed contract with the university", "call-2026-06-02") + + +def test_wrong_source_id_is_not_grounded(): + c = _corpus() + assert not c.contains("re-acknowledged by the funders", "email-1") + + +def test_too_short_quote_is_not_grounding(): + c = _corpus() + assert not c.contains("the funders", "call-2026-06-02") # 2 words + assert not SourceCorpus.is_substantive("at risk") + + +def test_duplicate_source_id_rejected(): + with pytest.raises(ValueError): + SourceCorpus.from_sources( + [Source(id="a", kind="email", text="x y z one two three"), + Source(id="a", kind="email", text="q")] + ) + + +def test_case_differences_are_tolerated(): + c = _corpus() + # sentence-start capitalization / all-caps emphasis in the retyped quote + assert c.contains("Re-acknowledged By The Funders before the autumn review", "call-2026-06-02") + + +def test_smart_quotes_and_dashes_are_folded(): + src = "Client: it’s about “value outward” — nothing less, we said." + c = SourceCorpus.from_sources([Source(id="s", kind="email", text=src)]) + # quote retyped with straight ASCII punctuation still grounds + assert c.contains('it\'s about "value outward" - nothing less', "s") + + +def test_invention_still_rejected_after_folding(): + c = _corpus() + # widening normalization must NOT let a fabricated claim through + assert not c.contains("we signed a five year exclusive with the university", "call-2026-06-02") + + +def test_contains_anywhere_finds_the_source(): + c = _corpus() + assert c.contains_anywhere("articulate the value outward") == "call-2026-06-02" + assert c.contains_anywhere("blockchain tokenomics moonshot") is None