interactive-story/backend/app/narrative/extract.py

"""M5: getting a typed proposal out of a narration, and keeping it out of the prose.

The model writes the story and, after it, one fenced block of typed events. This
module holds the instruction it is given, the parser that survives the ways a
model gets a format wrong, and the separation that keeps machine-readable output
from reaching the reader.

Two properties matter more than elegance here:

* **The prose must never carry the protocol.** A reader should not see a JSON
  block under their story, and a stored narration should not contain one either,
  because everything downstream — memory, summaries, export, the transcript —
  treats stored text as the story. The block is removed before the text is
  stored, not before it is displayed.
* **An unreadable block must not be a failed turn.** A narration the user watched
  arrive is worth keeping even when the state block after it is garbage. Parsing
  returns "no events" rather than raising, the turn commits with the state
  unchanged, and the proposal record keeps the raw output so the failure is
  visible in the audit rather than only in a log.
"""

from __future__ import annotations

import json
import re

from . import events

# The block the model is asked to append. Built from the vocabulary rather than
# written beside it, so the instruction cannot describe an event the application
# would then reject (`events.vocabulary_for_prompt`).
EMIT_RULE = (
    "After your narration, append a fenced code block labelled `state` containing "
    "a JSON object with an \"events\" list, recording what your own narration made "
    "true. Treat your narration as authoritative: if you wrote that someone moved, "
    "took something, learned something, was hurt, or that a new person or place "
    "appeared, record it.\n"
    "\n"
    "Every value is ABSOLUTE — the new state of things, never a change or a "
    "difference. Use only these events:\n"
    f"{events.vocabulary_for_prompt()}\n"
    "\n"
    "Identifiers are short lower-case slugs (mara, silver-key, old-abbey) and must "
    "match the ones already in the state you were shown. Introduce a person, place "
    "or thing with create_entity before referring to it. If the turn established "
    "nothing, send an empty events list.\n"
    "Example:\n"
    '```state\n'
    '{"events": [{"type": "set_possession", "item": "silver-key", "owner": "aldric"},'
    ' {"type": "set_current_location", "entity": "aldric", "location": "old-abbey"}]}\n'
    '```'
)

# Placed last, where recency is strongest, the same way the delta protocol did.
EMIT_REMINDER = (
    "[Reminder: end your reply with a ```state block listing the events your "
    "narration made true, with absolute values. Send an empty events list if "
    "nothing changed.]"
)

# Three patterns, and the difference between them is the whole of this module's
# safety. A story is allowed to contain code, and taking a code block out of
# someone's prose is a worse failure than leaving a stray proposal in it.
#
# `state` is the label the application asks for, so a fence carrying it is ours
# whatever is inside it — including a truncated `{oh no` that no JSON parser
# will take. That block must still leave the prose, and must still be recorded,
# because an unparseable proposal is exactly the failure the audit exists to
# make visible.
_STATE_FENCE_RE = re.compile(
    r"```state[^\S\n]*\n?(.*?)```", re.DOTALL | re.IGNORECASE
)
# `json` is *not* our label. Models reach for it anyway, so a ```json fence is
# taken only when what it contains is actually a proposal. A character who
# writes `{"name": "Mara"}` into a terminal keeps their code block (M5 review,
# Finding 6).
_JSON_FENCE_RE = re.compile(
    r"```json[^\S\n]*\n?(.*?)```", re.DOTALL | re.IGNORECASE
)
# An *unlabelled* fence is ours on the same terms: it has to be a proposal, not
# merely JSON-shaped.
_BARE_FENCE_RE = re.compile(r"```\s*([\[{].*?[\]}])\s*```", re.DOTALL)

# A bare object hugging the end of the text, for a model that forgets the fence.
_TRAILING_RE = re.compile(r"(\{.*\})\s*$", re.DOTALL)

# An opener with no closing fence. A model that runs out of output tokens
# mid-block leaves one of these, and everything after it is protocol rather than
# story — so the story ends where the opener begins.
#
# Our own label ends the story unconditionally. A dangling ```json fence is
# judged on what follows it, because an unterminated code block in a story is
# still the author's (M5 review, Finding 6).
_DANGLING_STATE_RE = re.compile(r"\n?```state\b.*\Z", re.DOTALL | re.IGNORECASE)
_DANGLING_JSON_RE = re.compile(r"\n?```json\b(.*)\Z", re.DOTALL | re.IGNORECASE)

# The reminder, parroted back. Small local models reproduce the bracketed
# instruction they were given, and it arrives as ordinary prose — no fence, so
# nothing above strips it, and the reader is shown a piece of the prompt.
#
# The bracket is *found* broadly and *judged* narrowly. Merely naming the
# protocol is not enough: a story may end on an aside about a state block, and
# deleting that sentence is the worse failure (M5 review, Finding 6). What marks
# the echo is the shape of the instruction itself — the fence token, the word it
# opens with, or the pair of phrases the reminder uses together.
_TRAILING_BRACKET_RE = re.compile(r"\n?\[([^\]]*)\]\s*\Z", re.DOTALL)


def _is_echoed_instruction(inner: str) -> bool:
    """Whether a trailing bracketed segment is the prompt's own reminder."""
    low = inner.lower()
    if "```state" in low:
        return True
    if low.lstrip().startswith("reminder:"):
        return True
    # The reminder names both; prose about the protocol rarely names either the
    # way the instruction does, and effectively never both.
    return "state block" in low and "events list" in low


def _clean(prose: str) -> str:
    """Removes protocol the block extraction could not, and nothing else.

    Found by the M5 realistic-context run (§12), which is the failure class
    Phase 0B warned about: under a full prompt the model echoed its own
    instruction into the narration, and the reader would have been shown it.
    Neither case here is hypothetical — both were observed against a real local
    model.
    """
    cleaned = prose
    bracket = _TRAILING_BRACKET_RE.search(cleaned)
    if bracket is not None and _is_echoed_instruction(bracket.group(1)):
        cleaned = cleaned[: bracket.start()]
    cleaned = _DANGLING_STATE_RE.sub("", cleaned)
    dangling = _DANGLING_JSON_RE.search(cleaned)
    if dangling is not None and _reads_as_protocol(dangling.group(1)):
        cleaned = cleaned[: dangling.start()]
    return cleaned.strip()


def _reads_as_protocol(tail: str) -> bool:
    """Whether a truncated fence was on its way to being a proposal."""
    if '"events"' in tail:
        return True
    return any(f'"{name}"' in tail for name in events.SPECS)


def _tolerant_load(blob: str):
    """Parses a block, forgiving what small local models get wrong.

    Trailing commas and a leading `+` on a number are both common and both
    rejected by strict JSON. Repairing them is not guessing at meaning — the
    intended value is unambiguous — which is the line this function stays on the
    right side of. Anything it cannot parse returns None, and the caller treats
    that as no proposal rather than as an empty one.
    """
    cleaned = re.sub(r",(\s*[}\]])", r"\1", blob)
    cleaned = re.sub(r"(:\s*)\+(\d)", r"\1\2", cleaned)
    try:
        parsed = json.loads(cleaned)
    except (json.JSONDecodeError, ValueError):
        return None
    return parsed


def split(text: str) -> tuple[str, dict | None, str]:
    """Separates a reply into `(prose, proposal, raw_block)`.

    `proposal` is None when there is no block or it cannot be parsed at all,
    which the caller records as a malformed proposal. `raw_block` is what the
    model actually wrote, kept for the audit record even — especially — when it
    did not parse.

    A bare trailing object is only stripped when it parses *and* looks like a
    proposal. Prose that happens to end in a brace is left alone, because
    removing a sentence from someone's story to satisfy a regex is a worse
    failure than leaving a stray brace in it.
    """
    matches = list(_STATE_FENCE_RE.finditer(text))
    if matches:
        match = matches[-1]
        raw = match.group(1).strip()
        prose = _clean(text[: match.start()] + text[match.end():])
        return prose, _tolerant_load(raw), raw

    # A `json` or unlabelled fence is ours only when its contents are this
    # protocol. That is judged two ways, and it needs both: a block that parses
    # into a proposal, or one that plainly reads as protocol even though it does
    # not parse. The second half matters — a small model that mangles its own
    # JSON must not have the wreckage shown to the reader, which is what the
    # realistic-model run caught during the corrective pass.
    for pattern in (_JSON_FENCE_RE, _BARE_FENCE_RE):
        for match in reversed(list(pattern.finditer(text))):
            raw = match.group(1).strip()
            parsed = _tolerant_load(raw)
            if _looks_like_proposal(parsed) or _reads_as_protocol(raw):
                prose = _clean(text[: match.start()] + text[match.end():])
                return prose, parsed, raw

    match = _TRAILING_RE.search(text)
    if match:
        raw = match.group(1)
        parsed = _tolerant_load(raw)
        if _looks_like_proposal(parsed):
            return _clean(text[: match.start()]), parsed, raw

    # No block at all — but the reply may still carry protocol the model wrote
    # as prose, or a fence it never closed.
    cleaned = _clean(text)
    if cleaned != text.strip():
        return cleaned, None, text.strip()[len(cleaned):].strip()
    return cleaned, None, ""


def _looks_like_proposal(parsed) -> bool:
    """Whether a bare trailing object is this protocol rather than prose."""
    if not isinstance(parsed, dict):
        return False
    if isinstance(parsed.get("events"), list):
        return True
    return isinstance(parsed.get("type"), str) and events.is_allowed(parsed["type"])


def render_block(accepted: list[dict]) -> str:
    """Renders accepted events back into the block the model emitted.

    Replayed into the prompt for past turns so the model copies the format it is
    being asked for. **Accepted** events rather than proposed ones, for the
    reason the delta protocol learned the hard way: showing the model a refused
    event standing as though it had worked, contradicted by the state in the
    same prompt, teaches it to send the event again.
    """
    if not accepted:
        return ""
    return "```state\n" + json.dumps({"events": accepted}, ensure_ascii=False) + "\n```"


def render_rejections(rejected: list[dict]) -> str:
    """The correction note appended after the most recent AI turn.

    Only what was lost. A model that is told what it got wrong can fix it next
    turn; a model told nothing repeats it.
    """
    if not rejected:
        return ""
    lines = []
    for entry in rejected[:6]:
        if not isinstance(entry, dict):
            continue
        detail = entry.get("detail") or entry.get("reason") or ""
        if detail:
            lines.append(f"- {detail}")
    if not lines:
        return ""
    body = "\n".join(lines)
    return (
        "[Part of your last state block was not accepted. Correct it in this "
        f"turn's block:\n{body}]"
    )