JesseMarkowitzandClaude Opus 5
b7005e6fdd
Replaces AI-DnD's RPG relative-delta world state with the genre-neutral typed
narrative state of ADR 010: explicit, absolute, allowlisted events proposed by
the model, validated by the application, applied to one authoritative document,
and snapshotted per position so restore stays a row read.
This commit includes the corrective pass that followed the independent review
in planning/reports/M5-IMPLEMENTATION-REPORT.md. The invariant it exists to
hold is:
visible active transcript position == stored head == authoritative state
Narrator editing (D10, STORY-BRANCH-SEMANTICS §§14-15)
A narrator edit no longer rewrites a row. It returns to the state before the
turn, takes the reader's exact text as the accepted narration, re-derives the
state that text implies, and becomes a new active continuation — while the
original narration keeps its words, its live flag and its whole future as
retained history. At the tip the correction is another take; with story below
it, it forks. No new history machinery: this is the existing fork/take/head
path with the reader's text in place of a generated reply. The §14A refusal
is therefore gone for narrator turns, and remains only for player input.
Pre-M5 positions
Migration 88 backfills the empty narrative document onto every action written
before M5, and a missing snapshot now restores the empty document instead of
leaving the previous position's state standing. Restoring to an old Save
Point no longer leaves a later position's entities and facts on screen.
Narrator context
Replayed history carries prose only; the machine-readable block is no longer
reconstructed into past turns, where it contradicted the authoritative state
in the same prompt. A fact withdrawn by a manual correction is now named as
no longer true, with the reader's reason, rather than silently dropped.
Also
- state_changes joins the action-list bulk read, removing one query per row.
- Extraction takes only the application's own protocol payload: an ordinary
```json or ```python block in a story survives, and a mangled proposal
still does not reach the reader.
Planning: ADR 013 records the authoritative document shape; §§14-15/14A, D10,
C04 and BUILD-MILESTONES are updated to describe what exists. Debt is recorded
against M8 (scenario editor UX) and M9 (export of the audit trail).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PWU4gTfLYY6Qq9U7aa9Qw2
1561 lines
66 KiB
Python
1561 lines
66 KiB
Python
"""M5: the genre-neutral authoritative narrative state.
|
|
|
|
This file is the acceptance contract for the milestone that replaced AI-DnD's
|
|
relative-delta RPG world state. Its subject is one claim:
|
|
|
|
The application decides what is true, from explicit typed proposals it
|
|
validated, and it can say why at any position in the story.
|
|
|
|
The tests are grouped by the question they answer, and named for the acceptance
|
|
items they discharge — C01-C04, C06, H05, L01-L02, J01-J03 in
|
|
`planning/V1-ACCEPTANCE-TESTS.md`.
|
|
|
|
Two things this file deliberately does **not** do. It does not test the model:
|
|
every proposal here is scripted, because what is under test is what the
|
|
application does with a proposal, not whether a given model produces a good one
|
|
— that is `test_narrative_realistic.py`, which needs a real model and says so.
|
|
And it does not re-test M3/M4 history; the history suites do that, now
|
|
instrumented through this state model.
|
|
|
|
python -m pytest tests/test_narrative_state.py -v
|
|
"""
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi import Depends
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import auth, head, limits, models
|
|
from app.database import Base, SessionLocal, engine, get_db
|
|
from app.main import app
|
|
from app.narrative import apply as napply
|
|
from app.narrative import events as nevents
|
|
from app.narrative import extract, model, store, validate
|
|
from app.routers import adventures
|
|
|
|
from fakes import ScriptedProvider, state_block
|
|
|
|
|
|
# --------------------------------------------------------------- fixtures
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
Base.metadata.create_all(bind=engine)
|
|
setup = SessionLocal()
|
|
user = models.User(is_guest=False, email="state@example.com")
|
|
setup.add(user)
|
|
setup.flush()
|
|
setup.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="test-model"))
|
|
adv = models.Adventure(user_id=user.id, title="The Crooked Lantern")
|
|
setup.add(adv)
|
|
setup.flush()
|
|
setup.add(models.Action(adventure_id=adv.id, type="start", text="The road forks."))
|
|
setup.commit()
|
|
adv_id, user_id = adv.id, user.id
|
|
setup.close()
|
|
|
|
ScriptedProvider.replies = ["Nothing happens." + "\n" + state_block([])]
|
|
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
|
|
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
|
|
|
|
def _current_user(db=Depends(get_db)):
|
|
return db.get(models.User, user_id)
|
|
|
|
app.dependency_overrides[auth.get_current_user] = _current_user
|
|
c = TestClient(app)
|
|
c.adv_id = adv_id
|
|
try:
|
|
yield c
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
adventures.turns._active_turns.clear()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
def _play(client, text="look around", reply=None, events=None):
|
|
if events is not None:
|
|
ScriptedProvider.replies = [f"{reply or 'It happens.'}\n{state_block(events)}"]
|
|
elif reply is not None:
|
|
ScriptedProvider.replies = [reply]
|
|
r = client.post(f"/api/adventures/{client.adv_id}/actions",
|
|
json={"type": "do", "text": text})
|
|
assert r.status_code == 200, r.text
|
|
return r
|
|
|
|
|
|
def _state(client) -> dict:
|
|
r = client.get(f"/api/adventures/{client.adv_id}/state")
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def _document(client) -> dict:
|
|
return _state(client)["document"]
|
|
|
|
|
|
def _events(client) -> list:
|
|
r = client.get(f"/api/adventures/{client.adv_id}/state/events")
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def _correct(client, events, note=""):
|
|
return client.post(
|
|
f"/api/adventures/{client.adv_id}/state/corrections",
|
|
json={"events": events, "note": note},
|
|
)
|
|
|
|
|
|
def _ai_ids(client) -> set:
|
|
"""The ids of every accepted narration, whatever the head is doing."""
|
|
db = SessionLocal()
|
|
try:
|
|
return {
|
|
a.id for a in db.query(models.Action).filter_by(
|
|
adventure_id=client.adv_id, type="ai")
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _proposals(adv_id):
|
|
db = SessionLocal()
|
|
try:
|
|
return (
|
|
db.query(models.StateProposal)
|
|
.filter_by(adventure_id=adv_id)
|
|
.order_by(models.StateProposal.id)
|
|
.all()
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _canon(adv_id, canon):
|
|
db = SessionLocal()
|
|
try:
|
|
adv = db.get(models.Adventure, adv_id)
|
|
adv.campaign_canon = canon
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
#: The fantasy fixture the acceptance tests are written against — the Continuity
|
|
#: Test's cast, as typed events.
|
|
FANTASY = [
|
|
{"type": "create_entity", "entity": "aldric", "entity_type": "character",
|
|
"name": "Aldric"},
|
|
{"type": "create_entity", "entity": "mara", "entity_type": "character",
|
|
"name": "Mara"},
|
|
{"type": "create_entity", "entity": "silver-key", "entity_type": "item",
|
|
"name": "the silver key"},
|
|
{"type": "create_entity", "entity": "old-abbey", "entity_type": "location",
|
|
"name": "the Old Abbey"},
|
|
{"type": "set_possession", "item": "silver-key", "owner": "aldric"},
|
|
{"type": "set_current_location", "entity": "aldric", "location": "old-abbey"},
|
|
]
|
|
|
|
|
|
# ====================================================== the event vocabulary
|
|
|
|
def test_every_allowed_event_applies_and_nothing_else_is_dispatched(client):
|
|
"""Each event in the vocabulary does something, and the vocabulary is the
|
|
whole of what can be done. A spec with no applier would accept a proposal
|
|
and silently change nothing — the one failure mode that looks like success.
|
|
"""
|
|
_play(client, "set the scene", events=FANTASY)
|
|
_play(client, "more", events=[
|
|
{"type": "set_entity_status", "entity": "mara", "status": "injured"},
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "resolve",
|
|
"value": 3},
|
|
{"type": "set_entity_conditions", "entity": "aldric",
|
|
"conditions": ["exhausted"]},
|
|
{"type": "add_fact", "subject": "mara", "predicate": "knows",
|
|
"object": "silver-key", "fact_id": "mara-knows-key"},
|
|
{"type": "add_relationship", "source": "aldric", "target": "mara",
|
|
"relationship": "trusts"},
|
|
{"type": "open_story_thread", "thread": "find-edrin",
|
|
"title": "Find out what happened to Edrin"},
|
|
{"type": "set_scene", "summary": "Rain on the abbey steps",
|
|
"location": "old-abbey"},
|
|
])
|
|
document = _document(client)
|
|
|
|
assert model.owner_of(document, "silver-key") == "aldric"
|
|
assert document["entities"]["aldric"]["location"] == "old-abbey"
|
|
assert document["entities"]["mara"]["status"] == "injured"
|
|
assert document["entities"]["mara"]["attributes"]["resolve"] == 3
|
|
assert document["entities"]["aldric"]["conditions"] == ["exhausted"]
|
|
assert model.knows(document, "mara", "silver-key")
|
|
assert any(r["type"] == "trusts" for r in model.active_relationships(document))
|
|
assert "find-edrin" in model.open_threads(document)
|
|
assert document["scene"]["location"] == "old-abbey"
|
|
|
|
# And the closing halves.
|
|
_play(client, "later", events=[
|
|
{"type": "clear_possession", "item": "silver-key"},
|
|
{"type": "invalidate_fact", "fact_id": "mara-knows-key"},
|
|
{"type": "end_relationship", "source": "aldric", "target": "mara",
|
|
"relationship": "trusts"},
|
|
{"type": "resolve_story_thread", "thread": "find-edrin"},
|
|
])
|
|
document = _document(client)
|
|
assert model.owner_of(document, "silver-key") is None
|
|
assert not model.knows(document, "mara", "silver-key")
|
|
assert model.active_relationships(document) == []
|
|
assert model.open_threads(document) == {}
|
|
# Withdrawn, not deleted: the record of what the campaign used to believe is
|
|
# what makes a correction auditable.
|
|
assert any(f["id"] == "mara-knows-key" for f in document["facts"])
|
|
|
|
|
|
def test_the_applier_covers_every_event_in_the_allowlist():
|
|
"""A spec with no branch in `apply` would be accepted and do nothing."""
|
|
state = model.empty()
|
|
state["entities"]["a"] = model.new_entity(name="A")
|
|
state["entities"]["b"] = model.new_entity(name="B")
|
|
state["facts"].append({"id": "f1", "predicate": "p", "status": "active"})
|
|
state["threads"]["t"] = {"title": "T", "status": "open"}
|
|
# `clear_possession` needs something to clear, and `end_relationship` an
|
|
# active tie: an event that correctly does nothing to an empty document
|
|
# would look identical to one with no applier at all.
|
|
state["entities"]["d"] = model.new_entity(name="D")
|
|
state["possessions"]["d"] = "b"
|
|
state["relationships"].append(
|
|
{"id": "r1", "source": "a", "target": "b", "type": "trusts",
|
|
"status": "active"})
|
|
|
|
samples = {
|
|
"create_entity": {"entity": "c", "name": "C"},
|
|
"set_entity_status": {"entity": "a", "status": "gone"},
|
|
"set_entity_attribute": {"entity": "a", "attribute": "x", "value": 1},
|
|
"set_entity_conditions": {"entity": "a", "conditions": ["hurt"]},
|
|
"set_current_location": {"entity": "a", "location": "b"},
|
|
"set_possession": {"item": "a", "owner": "b"},
|
|
"clear_possession": {"item": "d"},
|
|
"add_fact": {"predicate": "knows"},
|
|
"invalidate_fact": {"fact_id": "f1"},
|
|
"add_relationship": {"source": "a", "target": "b", "relationship": "trusts"},
|
|
"end_relationship": {"source": "a", "target": "b", "relationship": "trusts"},
|
|
"open_story_thread": {"thread": "t2", "title": "T2"},
|
|
"resolve_story_thread": {"thread": "t"},
|
|
"set_scene": {"summary": "here"},
|
|
}
|
|
assert set(samples) == set(nevents.ALLOWED), (
|
|
"the allowlist and this test's coverage have drifted apart"
|
|
)
|
|
for kind, payload in samples.items():
|
|
before = json.dumps(state, sort_keys=True)
|
|
after = napply.apply_events(state, [{"type": kind, **payload}])
|
|
assert json.dumps(after, sort_keys=True) != before, f"{kind} changed nothing"
|
|
|
|
|
|
# ============================================== H05 and the validation layers
|
|
|
|
def test_h05_execute_shell_is_rejected(client):
|
|
"""H05. The allowlist is checked before any field is read, so `command` is
|
|
never looked at — there is no branch in this application that could reach
|
|
it. The event is refused because it is not in the vocabulary, not because
|
|
"shell" is recognised."""
|
|
review = validate.review(
|
|
{"events": [{"event_type": "execute_shell", "command": "rm -rf /"}]},
|
|
model.empty(),
|
|
)
|
|
assert review.accepted == []
|
|
assert review.status == "rejected"
|
|
assert review.rejected[0].reason == validate.UNKNOWN_TYPE
|
|
|
|
# And end to end: a narration carrying it changes nothing and is recorded.
|
|
_play(client, "try it", events=[{"event_type": "execute_shell", "command": "id"}])
|
|
assert model.is_empty(_document(client))
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "rejected"
|
|
|
|
|
|
@pytest.mark.parametrize("payload, reason", [
|
|
({"type": "not_a_real_event", "entity": "a"}, validate.UNKNOWN_TYPE),
|
|
({"type": "set_entity_status"}, validate.MISSING_FIELD),
|
|
({"type": "set_entity_status", "entity": "ghost", "status": "x"},
|
|
validate.UNKNOWN_REFERENCE),
|
|
({"type": "set_entity_attribute", "entity": "a", "attribute": "x",
|
|
"value": {"nested": "structure"}}, validate.BAD_FIELD_TYPE),
|
|
({"type": "set_entity_conditions", "entity": "a", "conditions": "not a list"},
|
|
validate.BAD_FIELD_TYPE),
|
|
({"type": "create_entity", "entity": "a", "name": "duplicate"},
|
|
validate.DUPLICATE_ENTITY),
|
|
({"type": "set_possession", "item": "a", "owner": "a"},
|
|
validate.SELF_CONTRADICTION),
|
|
({"type": "invalidate_fact", "fact_id": "nope"}, validate.UNKNOWN_REFERENCE),
|
|
({"type": "resolve_story_thread", "thread": "nope"}, validate.UNKNOWN_REFERENCE),
|
|
("not an object", validate.NOT_AN_OBJECT),
|
|
])
|
|
def test_the_validator_refuses_each_class_of_bad_event(payload, reason):
|
|
state = model.empty()
|
|
state["entities"]["a"] = model.new_entity(name="A")
|
|
review = validate.review({"events": [payload]}, state)
|
|
assert review.accepted == []
|
|
assert review.rejected[0].reason == reason, review.rejected[0].as_dict()
|
|
|
|
|
|
@pytest.mark.parametrize("raw", [
|
|
"Just prose, no block at all.",
|
|
"Prose.\n```state\n{not json at all}\n```",
|
|
"Prose.\n```state\n[1, 2, 3]\n```",
|
|
'Prose.\n```state\n{"events": "not a list"}\n```',
|
|
'Prose.\n```state\n{"events": [null, 3, "x"]}\n```',
|
|
])
|
|
def test_malformed_proposals_never_mutate_state(client, raw):
|
|
"""A narration the user watched arrive is worth keeping even when the block
|
|
after it is garbage. The turn commits, the prose is stored clean, and the
|
|
state is untouched."""
|
|
_play(client, "establish", events=FANTASY)
|
|
before = _document(client)
|
|
|
|
_play(client, "then this", reply=raw)
|
|
|
|
assert _document(client) == before, "a malformed proposal changed the state"
|
|
# The prose survived, and carries no protocol.
|
|
texts = [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]]
|
|
assert "```state" not in "\n".join(texts)
|
|
assert "Prose." in texts[-1] or "Just prose" in texts[-1]
|
|
|
|
|
|
def test_one_bad_event_does_not_discard_the_good_ones(client):
|
|
"""Partial acceptance. Losing three correct events because the model
|
|
misspelled one entity would lose story the user watched happen."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "mixed", events=[
|
|
{"type": "set_entity_status", "entity": "mara", "status": "injured"},
|
|
{"type": "set_current_location", "entity": "nobody", "location": "old-abbey"},
|
|
{"type": "set_entity_conditions", "entity": "aldric", "conditions": ["cold"]},
|
|
])
|
|
document = _document(client)
|
|
assert document["entities"]["mara"]["status"] == "injured"
|
|
assert document["entities"]["aldric"]["conditions"] == ["cold"]
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "partially_accepted"
|
|
|
|
|
|
def test_a_proposal_may_introduce_an_entity_and_then_use_it(client):
|
|
"""Referential checks account for what earlier events in the same proposal
|
|
created — otherwise a narration could never introduce anyone."""
|
|
_play(client, "arrive", events=[
|
|
{"type": "create_entity", "entity": "edrin", "entity_type": "character",
|
|
"name": "Edrin"},
|
|
{"type": "create_entity", "entity": "cellar", "entity_type": "location",
|
|
"name": "the cellar"},
|
|
{"type": "set_current_location", "entity": "edrin", "location": "cellar"},
|
|
])
|
|
assert _document(client)["entities"]["edrin"]["location"] == "cellar"
|
|
|
|
|
|
def test_a_rejected_create_does_not_make_a_later_reference_resolve():
|
|
"""The running view must track what was *accepted*, not what was proposed."""
|
|
state = model.empty()
|
|
state["entities"]["mara"] = model.new_entity(name="Mara")
|
|
review = validate.review({"events": [
|
|
{"type": "create_entity", "entity": "mara", "name": "Duplicate"},
|
|
{"type": "create_entity", "entity": "hall", "name": "Hall"},
|
|
{"type": "set_current_location", "entity": "mara", "location": "hall"},
|
|
]}, state)
|
|
kinds = [e["type"] for e in review.accepted]
|
|
assert kinds == ["create_entity", "set_current_location"]
|
|
assert review.rejected[0].reason == validate.DUPLICATE_ENTITY
|
|
|
|
|
|
def test_a_proposal_cannot_carry_an_unbounded_number_of_events():
|
|
review = validate.review(
|
|
{"events": [{"type": "add_fact", "predicate": f"p{n}"} for n in range(80)]},
|
|
model.empty(),
|
|
)
|
|
assert len(review.accepted) == validate.MAX_EVENTS
|
|
assert review.rejected
|
|
|
|
|
|
# ================================================================ C01 canon
|
|
|
|
def test_c01_campaign_canon_outranks_the_narration(client):
|
|
"""C01, solved generically. The campaign declares the transition it forbids;
|
|
nothing in the application knows what resurrection is, and a science-fiction
|
|
campaign forbidding something else uses the same field and the same code."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she falls", events=[
|
|
{"type": "set_entity_status", "entity": "mara", "status": "dead"}])
|
|
assert _document(client)["entities"]["mara"]["status"] == "dead"
|
|
|
|
_canon(client.adv_id, {"forbidden_status_changes": [{"from": "dead", "to": "active"}]})
|
|
|
|
_play(client, "the rite", events=[
|
|
{"type": "set_entity_status", "entity": "mara", "status": "active"}])
|
|
|
|
assert _document(client)["entities"]["mara"]["status"] == "dead", (
|
|
"canon did not hold against the narration"
|
|
)
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "rejected"
|
|
|
|
|
|
def test_canon_reaches_the_prompt_as_well_as_the_validator(client):
|
|
"""C01 is a narration-time constraint too. Refusing the event after the
|
|
fact leaves the reader with prose the state contradicts; the model has to be
|
|
told the rule."""
|
|
_canon(client.adv_id, {"rules": ["The dead do not return."]})
|
|
_play(client, "go on", events=[])
|
|
db = SessionLocal()
|
|
try:
|
|
action = (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id, type="ai")
|
|
.order_by(models.Action.id.desc())
|
|
.first()
|
|
)
|
|
prompt = json.dumps(action.context_snapshot)
|
|
finally:
|
|
db.close()
|
|
assert "The dead do not return." in prompt
|
|
|
|
|
|
# ======================================================= C02 possession
|
|
|
|
def test_c02_possession_persists_until_an_event_changes_it(client):
|
|
_play(client, "establish", events=FANTASY)
|
|
for n in range(3):
|
|
_play(client, f"walk {n}", events=[])
|
|
assert model.owner_of(_document(client), "silver-key") == "aldric"
|
|
|
|
_play(client, "hand it over", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
assert model.owner_of(_document(client), "silver-key") == "mara"
|
|
|
|
|
|
def test_c02_undo_recovers_the_historically_correct_possession(client):
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "hand it over", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
assert model.owner_of(_document(client), "silver-key") == "mara"
|
|
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
assert model.owner_of(_document(client), "silver-key") == "aldric"
|
|
|
|
client.post(f"/api/adventures/{client.adv_id}/redo")
|
|
assert model.owner_of(_document(client), "silver-key") == "mara"
|
|
|
|
|
|
# ================================================== C03 character knowledge
|
|
|
|
def test_c03_what_the_campaign_knows_is_not_what_a_character_knows(client):
|
|
"""C03's distinction, made structural rather than inferred: a fact with no
|
|
subject is the campaign's; a fact whose subject is Mara is hers."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "the key's origin", events=[
|
|
{"type": "add_fact", "predicate": "was found in",
|
|
"subject": "silver-key", "object": "old-abbey", "fact_id": "key-origin"},
|
|
])
|
|
document = _document(client)
|
|
assert not model.knows(document, "mara", "key-origin"), (
|
|
"the campaign knowing something must not mean Mara knows it"
|
|
)
|
|
|
|
_play(client, "she is told", events=[
|
|
{"type": "add_fact", "subject": "mara", "predicate": "knows",
|
|
"object": "key-origin", "fact_id": "mara-learns"}])
|
|
assert model.knows(_document(client), "mara", "key-origin")
|
|
|
|
|
|
# ================================================ C04 manual correction
|
|
|
|
def test_c04_a_manual_correction_becomes_authoritative_and_is_auditable(client):
|
|
"""C04, end to end. The user overrules the story; the transcript is
|
|
untouched; the correction is marked as theirs and outranks the story."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she learns", events=[
|
|
{"type": "add_fact", "subject": "mara", "predicate": "knows",
|
|
"object": "silver-key", "fact_id": "mara-knows-key"}])
|
|
assert model.knows(_document(client), "mara", "silver-key")
|
|
transcript_before = [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]]
|
|
|
|
r = _correct(client, [
|
|
{"type": "invalidate_fact", "fact_id": "mara-knows-key",
|
|
"reason": "Mara never learned where the silver key was found."}],
|
|
note="Mara never learned where the silver key was found.")
|
|
assert r.status_code == 201, r.text
|
|
|
|
# ...the transcript is untouched by the correction itself...
|
|
assert [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]] == transcript_before
|
|
|
|
# ...it is authoritative for what follows...
|
|
assert not model.knows(_document(client), "mara", "silver-key")
|
|
_play(client, "carry on", events=[])
|
|
assert not model.knows(_document(client), "mara", "silver-key")
|
|
|
|
# ...auditable, and marked as the user's...
|
|
events = _events(client)
|
|
correction = next(e for e in events if e["source"] == "manual_correction")
|
|
assert correction["event_type"] == "invalidate_fact"
|
|
assert correction["before"]["status"] == "active"
|
|
assert correction["turn"] is not None
|
|
|
|
# ...and the story that was already written is still there, word for word,
|
|
# with the new turn appended rather than replacing anything.
|
|
after = [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]]
|
|
assert after[:len(transcript_before)] == transcript_before
|
|
|
|
|
|
def test_c04_a_correction_is_ranked_above_the_story_in_the_prompt(client):
|
|
_play(client, "establish", events=FANTASY)
|
|
_correct(client, [{"type": "add_fact", "subject": "mara",
|
|
"predicate": "never learned about", "object": "silver-key"}])
|
|
from app.narrative import render
|
|
rendered = render.for_prompt(_document(client))
|
|
assert "corrected by the player" in rendered
|
|
|
|
|
|
def test_c04_a_withdrawn_fact_does_not_come_back_through_history(client):
|
|
"""M5 review, Finding 4 — the review's own Mara example, as a regression.
|
|
|
|
The state section dropped the withdrawn fact and the history section handed
|
|
it straight back, replayed as an accepted `add_fact` in the protocol's own
|
|
words, with nothing in the prompt saying it had been corrected. The next
|
|
prompt therefore contradicted the correction it was supposed to carry.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she learns", events=[
|
|
{"type": "add_fact", "subject": "mara",
|
|
"predicate": "knows where the key was found", "fact_id": "mara-knows"}])
|
|
assert "knows where the key was found" in [
|
|
f["predicate"] for f in _document(client)["facts"]]
|
|
|
|
r = _correct(
|
|
client,
|
|
[{"type": "invalidate_fact", "fact_id": "mara-knows",
|
|
"reason": "Mara never learned where the silver key was found."}],
|
|
note="Mara never learned where the silver key was found.",
|
|
)
|
|
assert r.status_code in (200, 201), r.text
|
|
|
|
_play(client, "ask mara", events=[])
|
|
with SessionLocal() as db:
|
|
action = (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id, type="ai")
|
|
.order_by(models.Action.id.desc())
|
|
.first()
|
|
)
|
|
sections = {s["label"]: s["text"] for s in action.context_snapshot["sections"]}
|
|
|
|
# The history carries the story, and none of the protocol.
|
|
assert "```state" not in sections["history"]
|
|
assert "add_fact" not in sections["history"]
|
|
assert "mara-knows" not in sections["history"]
|
|
|
|
# The one place the assertion still appears says it is no longer true.
|
|
state = sections["narrative_state"]
|
|
assert "No longer true" in state
|
|
assert "Mara never learned where the silver key was found." in state
|
|
# And it is not standing among the facts that hold.
|
|
established = state.split("No longer true")[0]
|
|
assert "knows where the key was found" not in established
|
|
|
|
|
|
def test_a_withdrawn_fact_is_named_rather_than_silently_dropped(client):
|
|
"""Dropping it silently left the narration that first asserted it as the
|
|
only account in the prompt, and prose reads as current truth."""
|
|
from app.narrative import render
|
|
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she learns", events=[
|
|
{"type": "add_fact", "subject": "mara", "predicate": "knows the code",
|
|
"fact_id": "code"}])
|
|
_correct(client, [{"type": "invalidate_fact", "fact_id": "code",
|
|
"reason": "She was never told."}])
|
|
|
|
rendered = render.for_prompt(_document(client))
|
|
|
|
assert "No longer true — do not treat these as established:" in rendered
|
|
assert "knows the code" in rendered.split("No longer true")[1]
|
|
assert "She was never told." in rendered
|
|
|
|
|
|
def test_a_correction_goes_through_the_same_validator(client):
|
|
"""A user is trusted with authority, not with references that do not
|
|
resolve: a typo should be a clear refusal, not a corrupt document."""
|
|
_play(client, "establish", events=FANTASY)
|
|
r = _correct(client, [{"type": "set_entity_status", "entity": "nobody",
|
|
"status": "gone"}])
|
|
assert r.status_code == 400
|
|
assert "nobody" in r.json()["detail"]
|
|
|
|
r = _correct(client, [{"type": "execute_shell", "command": "id"}])
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_a_correction_belongs_to_the_position_it_was_made_at(client):
|
|
"""Undoing past a correction drops it, and redoing brings it back — the
|
|
same rule every other state change follows."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "second turn", events=[])
|
|
_correct(client, [{"type": "set_entity_status", "entity": "mara",
|
|
"status": "vanished"}])
|
|
assert _document(client)["entities"]["mara"]["status"] == "vanished"
|
|
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
assert _document(client)["entities"]["mara"]["status"] != "vanished"
|
|
|
|
client.post(f"/api/adventures/{client.adv_id}/redo")
|
|
assert _document(client)["entities"]["mara"]["status"] == "vanished"
|
|
|
|
|
|
# ================================================ C06 narration/state coherence
|
|
|
|
def test_c06_accepted_state_matches_the_accepted_narration(client):
|
|
"""C06 with an unambiguous consequence: the item moves in the prose, and the
|
|
state says it moved."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "hand it over",
|
|
reply="Aldric presses the silver key into Mara's palm.",
|
|
events=[{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
|
|
document = _document(client)
|
|
assert model.owner_of(document, "silver-key") == "mara"
|
|
texts = [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]]
|
|
assert "presses the silver key" in texts[-1]
|
|
assert "```state" not in texts[-1], "the protocol reached the reader"
|
|
|
|
|
|
def test_c06_a_contradictory_proposal_is_refused_rather_than_accepted(client):
|
|
"""Two events in one proposal that cannot both be true of the same item."""
|
|
_play(client, "establish", events=FANTASY)
|
|
review = validate.review({"events": [
|
|
{"type": "set_possession", "item": "silver-key", "owner": "silver-key"},
|
|
]}, _document(client))
|
|
assert review.accepted == []
|
|
assert review.rejected[0].reason == validate.SELF_CONTRADICTION
|
|
|
|
|
|
# =========================================================== §8 provenance
|
|
|
|
def test_every_accepted_event_records_what_it_changed_and_why(client):
|
|
"""§8's list: what changed, which turn, model or user, and what it was."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "move", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
|
|
events = _events(client)
|
|
move = next(e for e in events if e["event_type"] == "set_possession")
|
|
assert move["payload"]["owner"] == "mara"
|
|
assert move["before"] == {"owner": "aldric"}, "the previous value was not recorded"
|
|
assert move["source"] == "accepted_story"
|
|
assert move["action_id"] is not None
|
|
assert move["depth"] is not None
|
|
|
|
|
|
def test_a_rejected_proposal_is_recorded_but_is_not_an_event(client):
|
|
"""A rejection has to be inspectable — a wrong-looking campaign with no
|
|
trail is the failure this record exists to prevent — without becoming
|
|
authoritative."""
|
|
_play(client, "establish", events=FANTASY)
|
|
before = len(_events(client))
|
|
_play(client, "impossible", events=[
|
|
{"type": "set_current_location", "entity": "ghost", "location": "old-abbey"}])
|
|
|
|
assert len(_events(client)) == before, "a rejected event became authoritative"
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "rejected"
|
|
db = SessionLocal()
|
|
try:
|
|
detail = db.get(models.StateProposal, proposal.id).detail
|
|
finally:
|
|
db.close()
|
|
assert detail["rejected"][0]["reason"] == validate.UNKNOWN_REFERENCE
|
|
|
|
|
|
def test_an_unparseable_block_keeps_the_raw_output_for_the_audit(client):
|
|
"""The one case no structured column could hold."""
|
|
_play(client, "garbage", reply="Prose.\n```state\n{oh no\n```")
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "unparseable"
|
|
assert "oh no" in proposal.raw_output
|
|
|
|
|
|
# ================================================== L01 atomic turn commit
|
|
|
|
def test_l01_a_failed_state_commit_accepts_no_narration(client, monkeypatch):
|
|
"""L01. There must be no window in which narration is accepted while its
|
|
state is half-written, so the whole turn is one transaction."""
|
|
_play(client, "establish", events=FANTASY)
|
|
before_state = _document(client)
|
|
before_events = len(_events(client))
|
|
ai_before = _ai_ids(client)
|
|
|
|
def explode(*a, **k):
|
|
raise RuntimeError("state persistence failed")
|
|
|
|
monkeypatch.setattr(adventures.turns.narrative.store, "record", explode)
|
|
with pytest.raises(RuntimeError):
|
|
_play(client, "the turn that fails", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
|
|
# A05 deliberately keeps the player's submitted text so it can be tried
|
|
# again, so what L01 forbids is narrower and sharper: no *narration* was
|
|
# accepted, no state moved, and no event was recorded.
|
|
assert _ai_ids(client) == ai_before, "a narration was accepted"
|
|
assert _document(client) == before_state, "state moved for a turn that failed"
|
|
assert len(_events(client)) == before_events, "an event outlived its turn"
|
|
|
|
|
|
def test_l01_the_events_and_the_snapshot_land_with_the_turn(client):
|
|
"""The positive half: one commit, everything present after it."""
|
|
_play(client, "establish", events=FANTASY)
|
|
db = SessionLocal()
|
|
try:
|
|
action = (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id, type="ai")
|
|
.order_by(models.Action.id.desc())
|
|
.first()
|
|
)
|
|
assert isinstance(action.narrative_state_after, dict), "no snapshot"
|
|
assert model.owner_of(action.narrative_state_after, "silver-key") == "aldric"
|
|
assert db.query(models.StateEvent).filter_by(action_id=action.id).count() \
|
|
== len(FANTASY)
|
|
assert db.query(models.StateProposal).filter_by(action_id=action.id).count() == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ============================================ L02 positional reconstruction
|
|
|
|
def test_l02_state_at_every_restored_position_is_what_was_accepted_there(client):
|
|
"""L02, measured in both directions. The state at a position is the state
|
|
that position produced — arriving from in front of it and from behind it
|
|
must agree."""
|
|
_play(client, "establish", events=FANTASY)
|
|
owners = ["aldric"]
|
|
for owner in ("mara", "aldric", "mara"):
|
|
_play(client, f"pass to {owner}", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": owner}])
|
|
owners.append(owner)
|
|
|
|
going_back = []
|
|
for _ in range(len(owners) - 1):
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
going_back.append(model.owner_of(_document(client), "silver-key"))
|
|
|
|
coming_forward = []
|
|
for _ in range(len(owners) - 1):
|
|
client.post(f"/api/adventures/{client.adv_id}/redo")
|
|
coming_forward.append(model.owner_of(_document(client), "silver-key"))
|
|
|
|
assert going_back == list(reversed(owners[:-1]))
|
|
assert coming_forward == owners[1:]
|
|
|
|
|
|
def test_restoring_a_position_is_a_row_read_not_a_replay(client):
|
|
"""The performance property M3/M4 made load-bearing
|
|
(`TECHNICAL-DESIGN.md` §10.4): a restore must not scale with the campaign.
|
|
|
|
Asserted on query count rather than on time, because a timing assertion in a
|
|
test suite is a flake waiting to happen. An event-replay implementation
|
|
would have to read the event log, and the count would grow with the story.
|
|
"""
|
|
from sqlalchemy import event as sa_event
|
|
|
|
_play(client, "establish", events=FANTASY)
|
|
for n in range(8):
|
|
_play(client, f"turn {n}", events=[
|
|
{"type": "set_entity_attribute", "entity": "aldric",
|
|
"attribute": "steps", "value": n}])
|
|
|
|
seen = []
|
|
|
|
def record(conn, cursor, statement, params, context, executemany):
|
|
seen.append(statement)
|
|
|
|
sa_event.listen(engine, "before_cursor_execute", record)
|
|
try:
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
finally:
|
|
sa_event.remove(engine, "before_cursor_execute", record)
|
|
|
|
assert not any("state_events" in q for q in seen), (
|
|
"restoring read the event log — that is a replay, not a snapshot"
|
|
)
|
|
|
|
|
|
# ============================================ J01-J03 genre neutrality
|
|
|
|
SCIFI = [
|
|
{"type": "create_entity", "entity": "persephone", "entity_type": "vehicle",
|
|
"name": "the Persephone"},
|
|
{"type": "create_entity", "entity": "ceres-station", "entity_type": "location",
|
|
"name": "Ceres Station"},
|
|
{"type": "create_entity", "entity": "helios-combine",
|
|
"entity_type": "organization", "name": "the Helios Combine"},
|
|
{"type": "create_entity", "entity": "data-crystal", "entity_type": "item",
|
|
"name": "the encrypted data crystal"},
|
|
{"type": "create_entity", "entity": "vela", "entity_type": "character",
|
|
"name": "Vela"},
|
|
{"type": "set_current_location", "entity": "persephone",
|
|
"location": "ceres-station"},
|
|
{"type": "set_possession", "item": "data-crystal", "owner": "vela"},
|
|
{"type": "add_fact", "predicate": "cannot exceed light speed",
|
|
"subject": "persephone", "fact_id": "no-ftl"},
|
|
{"type": "add_relationship", "source": "vela", "target": "helios-combine",
|
|
"relationship": "works for"},
|
|
{"type": "open_story_thread", "thread": "decrypt-the-crystal",
|
|
"title": "Decrypt the data crystal"},
|
|
]
|
|
|
|
|
|
def test_j01_j02_a_science_fiction_campaign_uses_the_same_schema(client):
|
|
"""J01 and J02. A ship, a corporation, a station and a data crystal, with no
|
|
schema change, no new table and no genre-specific branch."""
|
|
_play(client, "dock", events=SCIFI)
|
|
document = _document(client)
|
|
|
|
assert document["entities"]["persephone"]["type"] == "vehicle"
|
|
assert document["entities"]["helios-combine"]["type"] == "organization"
|
|
assert document["entities"]["ceres-station"]["type"] == "location"
|
|
assert document["entities"]["data-crystal"]["type"] == "item"
|
|
assert document["entities"]["persephone"]["location"] == "ceres-station"
|
|
assert model.owner_of(document, "data-crystal") == "vela"
|
|
assert "decrypt-the-crystal" in model.open_threads(document)
|
|
|
|
|
|
def test_j03_only_the_data_differs_between_the_two_genres(client):
|
|
"""J03. The two fixtures exercise the identical event vocabulary, and the
|
|
documents they produce have identical structure — the difference is names."""
|
|
fantasy = napply.apply_events(model.empty(), FANTASY)
|
|
scifi = napply.apply_events(model.empty(), SCIFI)
|
|
|
|
def shape(document):
|
|
return {
|
|
key: type(value).__name__ for key, value in document.items()
|
|
}
|
|
|
|
assert shape(fantasy) == shape(scifi)
|
|
assert {e["type"] for e in FANTASY} <= set(nevents.ALLOWED)
|
|
assert {e["type"] for e in SCIFI} <= set(nevents.ALLOWED)
|
|
# And nothing in the application names a genre.
|
|
from pathlib import Path
|
|
package = Path(__file__).resolve().parents[1] / "app" / "narrative"
|
|
source = "\n".join(p.read_text(encoding="utf-8") for p in package.glob("*.py"))
|
|
import re as _re
|
|
for word in ("resurrect", "magic", "sword", "hit point", "spaceship",
|
|
"faster-than-light", "spell", "armor", "mana", "dungeon"):
|
|
# Word boundaries, because "misspelled" is not a genre assumption.
|
|
assert not _re.search(rf"\b{_re.escape(word)}", source.lower()), (
|
|
f"the state engine names {word!r}"
|
|
)
|
|
|
|
|
|
def test_j03_the_inspector_labels_categories_it_was_not_taught(client):
|
|
"""A campaign inventing its own entity category must not fall into "Other"."""
|
|
_play(client, "dock", events=SCIFI + [
|
|
{"type": "create_entity", "entity": "the-drift", "entity_type": "phenomenon",
|
|
"name": "the Drift"}])
|
|
titles = [g["title"] for g in _state(client)["groups"]]
|
|
assert "Vehicles" in titles
|
|
assert "Organizations" in titles
|
|
assert any(t.lower().startswith("phenomenon") for t in titles), titles
|
|
|
|
|
|
# ============================================== the inspector and the prompt
|
|
|
|
def test_the_inspector_shows_only_what_the_campaign_has(client):
|
|
empty = _state(client)
|
|
assert empty["empty"] is True and empty["groups"] == []
|
|
|
|
_play(client, "establish", events=FANTASY)
|
|
view = _state(client)
|
|
titles = [g["title"] for g in view["groups"]]
|
|
assert "Characters" in titles and "Locations" in titles
|
|
assert "Relationships" not in titles, "a category with nothing in it was shown"
|
|
rows = next(g for g in view["groups"] if g["title"] == "Characters")["rows"]
|
|
assert any(r["key"] == "aldric" and "Old Abbey" in r["detail"] for r in rows)
|
|
|
|
|
|
def test_the_prompt_carries_the_state_and_the_vocabulary(client):
|
|
_play(client, "establish", events=FANTASY)
|
|
db = SessionLocal()
|
|
try:
|
|
action = (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id, type="ai")
|
|
.order_by(models.Action.id.desc())
|
|
.first()
|
|
)
|
|
snapshot = action.context_snapshot
|
|
finally:
|
|
db.close()
|
|
prompt = json.dumps(snapshot)
|
|
assert "set_possession" in prompt, "the model was not told the vocabulary"
|
|
assert "ABSOLUTE" in prompt, "the model was not told values are absolute"
|
|
|
|
|
|
def test_the_emit_rule_only_describes_events_that_exist():
|
|
"""Generated from the allowlist, so the instruction cannot drift into
|
|
describing an event the application would then reject."""
|
|
rule = extract.EMIT_RULE
|
|
for name in nevents.ALLOWED:
|
|
assert name in rule
|
|
assert "increment" not in rule.lower(), "an ambiguous operation was described"
|
|
|
|
|
|
# ================================================== extraction and the prose
|
|
|
|
@pytest.mark.parametrize("reply, prose", [
|
|
('Story.\n```state\n{"events": []}\n```', "Story."),
|
|
('Story.\n```json\n{"events": []}\n```', "Story."),
|
|
('Story.\n```\n{"events": []}\n```', "Story."),
|
|
('Story.\n{"events": []}', "Story."),
|
|
('Story with a brace }', "Story with a brace }"),
|
|
])
|
|
def test_the_block_is_separated_from_the_prose(reply, prose):
|
|
text, _parsed, _raw = extract.split(reply)
|
|
assert text == prose
|
|
|
|
|
|
def test_trailing_prose_that_is_not_a_proposal_is_left_alone():
|
|
"""Removing a sentence from someone's story to satisfy a regex is worse
|
|
than leaving a stray brace in it."""
|
|
reply = 'She said, "the vault is sealed {for now}"'
|
|
text, parsed, _raw = extract.split(reply)
|
|
assert text == reply
|
|
assert parsed is None
|
|
|
|
|
|
def test_tolerant_parsing_repairs_only_unambiguous_mistakes():
|
|
text, parsed, _ = extract.split(
|
|
'Story.\n```state\n{"events": [{"type": "add_fact", "predicate": "x",}],}\n```')
|
|
assert parsed == {"events": [{"type": "add_fact", "predicate": "x"}]}
|
|
|
|
|
|
# ================================ D10 / §14-15: the narrator edit re-evaluates
|
|
|
|
def _edit(client, action_id, text):
|
|
return client.patch(
|
|
f"/api/adventures/{client.adv_id}/actions/{action_id}",
|
|
json={"text": text},
|
|
)
|
|
|
|
|
|
def _last_ai(client):
|
|
db = SessionLocal()
|
|
try:
|
|
return (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id, type="ai")
|
|
.order_by(models.Action.id.desc())
|
|
.first()
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_d10_editing_a_narration_re_derives_its_state(client):
|
|
"""`STORY-BRANCH-SEMANTICS.md` §15, and the half of D10 M5 owns.
|
|
|
|
The invariant, exactly: corrected authoritative prose and authoritative
|
|
structured state must not disagree because the edit changed only the text.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she dresses",
|
|
reply="Mara pulls on a red cloak.",
|
|
events=[{"type": "set_entity_attribute", "entity": "mara",
|
|
"attribute": "clothing", "value": "red cloak"}])
|
|
assert _document(client)["entities"]["mara"]["attributes"]["clothing"] == "red cloak"
|
|
|
|
edited = _last_ai(client)
|
|
r = _edit(client, edited.id,
|
|
'Mara pulls on a green cloak.\n' + state_block([
|
|
{"type": "set_entity_attribute", "entity": "mara",
|
|
"attribute": "clothing", "value": "green cloak"}]))
|
|
assert r.status_code == 200, r.text
|
|
|
|
assert _document(client)["entities"]["mara"]["attributes"]["clothing"] == "green cloak"
|
|
# The block the user pasted in does not become part of the story.
|
|
assert "```state" not in r.json()["text"]
|
|
assert "green cloak" in r.json()["text"]
|
|
|
|
|
|
def test_an_edit_re_derives_from_the_state_before_the_turn(client):
|
|
"""Not from the campaign's current state. An edit replaces what this turn
|
|
established; it must not stack on top of what the turn already did."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she moves",
|
|
events=[{"type": "set_current_location", "entity": "mara",
|
|
"location": "old-abbey"}])
|
|
assert _document(client)["entities"]["mara"]["location"] == "old-abbey"
|
|
|
|
# The edited narration says she stayed put, and proposes nothing.
|
|
edited = _last_ai(client)
|
|
_edit(client, edited.id, "Mara does not move.")
|
|
|
|
assert _document(client)["entities"]["mara"]["location"] is None, (
|
|
"the edit inherited the state its own turn had established"
|
|
)
|
|
|
|
|
|
def test_an_edit_leaves_the_audit_trail_of_both_versions(client):
|
|
"""Nothing is rewritten backwards: the events the original narration
|
|
produced stay in the log, and the correction's are appended beside them."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she dresses",
|
|
events=[{"type": "set_entity_attribute", "entity": "mara",
|
|
"attribute": "clothing", "value": "red cloak"}])
|
|
before = len(_events(client))
|
|
|
|
edited = _last_ai(client)
|
|
_edit(client, edited.id,
|
|
'Green.\n' + state_block([
|
|
{"type": "set_entity_attribute", "entity": "mara",
|
|
"attribute": "clothing", "value": "green cloak"}]))
|
|
|
|
events = _events(client)
|
|
assert len(events) > before, "the edit recorded nothing"
|
|
assert any(e["source"] == "narrator_edit" for e in events)
|
|
# Both readings are still on the record.
|
|
values = [e["payload"].get("value") for e in events
|
|
if e["event_type"] == "set_entity_attribute"]
|
|
assert "red cloak" in values and "green cloak" in values
|
|
|
|
|
|
def test_an_edit_whose_proposal_is_refused_establishes_nothing(client):
|
|
"""A refused proposal inside an edit must not leave a half-state behind.
|
|
|
|
The edited turn now establishes nothing — its narration replaced the one
|
|
that set the cloak, and what the new narration proposed was refused — so the
|
|
attribute is gone rather than left at either value. That is the coherent
|
|
outcome: prose and state agree that this turn established nothing, which is
|
|
the invariant, rather than the state keeping a claim no narration makes.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she dresses",
|
|
events=[{"type": "set_entity_attribute", "entity": "mara",
|
|
"attribute": "clothing", "value": "red cloak"}])
|
|
assert _document(client)["entities"]["mara"]["attributes"]["clothing"] == "red cloak"
|
|
|
|
edited = _last_ai(client)
|
|
r = _edit(client, edited.id,
|
|
'Green.\n' + state_block([
|
|
{"type": "set_entity_attribute", "entity": "nobody",
|
|
"attribute": "clothing", "value": "green cloak"}]))
|
|
assert r.status_code == 200, r.text
|
|
|
|
attributes = _document(client)["entities"]["mara"]["attributes"]
|
|
assert "clothing" not in attributes, (
|
|
"a claim survived the narration that made it"
|
|
)
|
|
assert "green cloak" not in str(_document(client)), (
|
|
"a refused value reached the state"
|
|
)
|
|
# And the refusal is on the record rather than silent.
|
|
proposal = _proposals(client.adv_id)[-1]
|
|
assert proposal.status == "rejected"
|
|
assert proposal.source == "narrator_edit"
|
|
|
|
|
|
# ------------------------------------------------------------------ D10 / §§14-15
|
|
#
|
|
# The M5 corrective pass. The review (Finding 1) found a narrator edit rewinding
|
|
# the campaign's live state to the edited turn while the head stayed at the tip,
|
|
# so the reader saw a full transcript over a state document describing an
|
|
# earlier moment. These tests hold the invariant that failure broke:
|
|
#
|
|
# visible active transcript position == stored head == authoritative state
|
|
|
|
|
|
def _texts(client) -> list[str]:
|
|
r = client.get(f"/api/adventures/{client.adv_id}")
|
|
assert r.status_code == 200, r.text
|
|
return [a["text"] for a in r.json()["actions"]]
|
|
|
|
|
|
def _head_of(client) -> tuple[int, int]:
|
|
with SessionLocal() as db:
|
|
adventure = db.get(models.Adventure, client.adv_id)
|
|
return adventure.head_branch_id, adventure.head_depth
|
|
|
|
|
|
def _live_state_matches_head_snapshot(client) -> bool:
|
|
"""The invariant, read straight out of the database."""
|
|
with SessionLocal() as db:
|
|
adventure = db.get(models.Adventure, client.adv_id)
|
|
# Through the lineage, not by branch id: after a fork the head branch
|
|
# owns one node and inherits the rest of the path from its parent.
|
|
node = head.node_at(db, adventure, adventure.head_depth)
|
|
assert node is not None, "the head rests on no node"
|
|
return model.normalize(adventure.narrative_state) == model.normalize(
|
|
node.narrative_state_after
|
|
)
|
|
|
|
|
|
def test_editing_a_narrator_turn_with_visible_descendants_forks(client):
|
|
"""The review's reproduction, as a regression.
|
|
|
|
Four turns, each establishing something, then the *earliest* narrator turn
|
|
is corrected while every later turn is still on screen. The correction has
|
|
to become a new continuation rather than a rewind of the line being read.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she dresses", events=[
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "cloak",
|
|
"value": "red"}])
|
|
_play(client, "walk on", events=[
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "boots",
|
|
"value": "muddy"}])
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][1]
|
|
target_id, original_text = target.id, target.text
|
|
descendants = [
|
|
a.id for a in _rows_all(client) if (a.depth or 0) > (target.depth or 0)
|
|
]
|
|
assert descendants, "the fixture needs visible story below the edited turn"
|
|
branch_before, _ = _head_of(client)
|
|
|
|
r = _edit(client, target_id, "Mara pulls on a green cloak.\n" + state_block([
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "cloak",
|
|
"value": "green"}]))
|
|
assert r.status_code == 200, r.text
|
|
|
|
rows = {a.id: a for a in _rows_all(client)}
|
|
# §15.5-6: nothing on the old line was written to.
|
|
assert rows[target_id].text == original_text
|
|
assert all(old_id in rows for old_id in descendants)
|
|
# §15.4: a new active continuation, and the head is on it.
|
|
branch_after, depth_after = _head_of(client)
|
|
assert branch_after != branch_before, "the correction did not fork"
|
|
assert depth_after == target.depth
|
|
# §15.1-3: state re-derived from the turn's own starting point.
|
|
mara = _document(client)["entities"]["mara"]["attributes"]
|
|
assert mara["cloak"] == "green"
|
|
assert "boots" not in mara, "state from the abandoned future stayed current"
|
|
# The invariant.
|
|
assert _live_state_matches_head_snapshot(client)
|
|
|
|
|
|
def test_the_corrected_text_is_used_verbatim_not_regenerated(client):
|
|
"""§15.2. The reader's words are the narration; no model is called."""
|
|
_play(client, "establish", events=FANTASY)
|
|
ScriptedProvider.replies = ["THE MODEL SHOULD NOT BE CALLED."]
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][0]
|
|
|
|
r = _edit(client, target.id, "Mara sets the key down, exactly so.")
|
|
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["text"] == "Mara sets the key down, exactly so."
|
|
assert "THE MODEL SHOULD NOT BE CALLED." not in _texts(client)
|
|
|
|
|
|
def test_the_old_narration_and_its_future_leave_the_active_transcript(client):
|
|
"""§15.4-5: retained, but not read. The old line keeps its rows; the story
|
|
on screen is the corrected one."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she learns", events=[
|
|
{"type": "add_fact", "subject": "mara", "predicate": "knows the code",
|
|
"fact_id": "code"}])
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][0]
|
|
|
|
_edit(client, target.id, "A different opening entirely.")
|
|
|
|
visible = _texts(client)
|
|
assert "A different opening entirely." in visible
|
|
assert not any("knows the code" in t for t in visible)
|
|
facts = [f["predicate"] for f in _document(client)["facts"]]
|
|
assert "knows the code" not in facts
|
|
assert _live_state_matches_head_snapshot(client)
|
|
|
|
|
|
def test_editing_the_latest_narrator_turn_still_works(client):
|
|
"""At the tip the correction is a take: nothing was built on the turn, so no
|
|
branch is needed, and the original stays in the pager (§15.5)."""
|
|
_play(client, "establish", events=FANTASY)
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][-1]
|
|
original_text = target.text
|
|
branch_before, depth_before = _head_of(client)
|
|
|
|
r = _edit(client, target.id, "Corrected at the tip.\n" + state_block([
|
|
{"type": "add_fact", "predicate": "the tip was corrected",
|
|
"fact_id": "tip"}]))
|
|
|
|
assert r.status_code == 200, r.text
|
|
assert _head_of(client) == (branch_before, depth_before), "no branch needed"
|
|
rows = {a.id: a for a in _rows_all(client)}
|
|
assert rows[target.id].text == original_text, "the original take is retained"
|
|
assert not rows[target.id].live
|
|
assert "Corrected at the tip." in _texts(client)
|
|
assert "the tip was corrected" in [f["predicate"] for f in _document(client)["facts"]]
|
|
assert _live_state_matches_head_snapshot(client)
|
|
|
|
|
|
def test_an_edit_is_safe_when_a_future_is_off_screen(client):
|
|
"""What §14A refused, §§14-15 handle.
|
|
|
|
The refusal existed because an in-place edit changed the words an invisible
|
|
stretch of story was written from. A fork writes nothing to that line, so
|
|
the case is no longer unsafe — it is ordinary.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "second", events=[])
|
|
_play(client, "third", events=[])
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][0]
|
|
original_text = target.text
|
|
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
off_screen = [
|
|
a.id for a in _rows_all(client) if (a.depth or 0) > (target.depth or 0)
|
|
]
|
|
|
|
r = _edit(client, target.id, "Something else entirely.")
|
|
|
|
assert r.status_code == 200, r.text
|
|
rows = {a.id: a for a in _rows_all(client)}
|
|
assert rows[target.id].text == original_text, "the off-screen line kept its words"
|
|
assert all(old_id in rows for old_id in off_screen), "and kept its story"
|
|
assert "Something else entirely." in _texts(client)
|
|
assert _live_state_matches_head_snapshot(client)
|
|
|
|
|
|
def test_undo_and_redo_after_an_edit_stay_on_the_corrected_lineage(client):
|
|
"""The consequence the review found worst: Undo restoring a snapshot from a
|
|
line the reader is no longer on, so the prose said green and the state said
|
|
red. Every position visited here has to agree with itself."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "she dresses", events=[
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "cloak",
|
|
"value": "red"}])
|
|
_play(client, "walk on", events=[
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "boots",
|
|
"value": "muddy"}])
|
|
target = [a for a in _rows_all(client) if a.type == "ai"][1]
|
|
|
|
_edit(client, target.id, "Mara pulls on a green cloak.\n" + state_block([
|
|
{"type": "set_entity_attribute", "entity": "mara", "attribute": "cloak",
|
|
"value": "green"}]))
|
|
|
|
assert _live_state_matches_head_snapshot(client)
|
|
for _ in range(2):
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
assert _live_state_matches_head_snapshot(client)
|
|
# Nothing from the abandoned line may reappear.
|
|
mara = _document(client)["entities"].get("mara", {}).get("attributes", {})
|
|
assert mara.get("cloak") != "red", "a snapshot from the old line was restored"
|
|
assert "boots" not in mara
|
|
for _ in range(2):
|
|
client.post(f"/api/adventures/{client.adv_id}/redo")
|
|
assert _live_state_matches_head_snapshot(client)
|
|
assert _document(client)["entities"]["mara"]["attributes"]["cloak"] == "green"
|
|
|
|
|
|
def _rows_all(client):
|
|
db = SessionLocal()
|
|
try:
|
|
return (
|
|
db.query(models.Action)
|
|
.filter_by(adventure_id=client.adv_id)
|
|
.order_by(models.Action.depth, models.Action.id)
|
|
.all()
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ============================================ export / import round trip
|
|
|
|
def test_the_bundle_carries_the_state_and_its_snapshots(client):
|
|
"""M5 must not make export silently lose the authoritative state.
|
|
|
|
Both halves matter. The campaign's document is what the imported story
|
|
believes; the per-node snapshots are what makes its history walkable, and a
|
|
bundle carrying the turns without them would import a story that reads
|
|
correctly and then restores to the wrong state.
|
|
"""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "hand it over", events=[
|
|
{"type": "set_possession", "item": "silver-key", "owner": "mara"}])
|
|
_canon(client.adv_id, {"rules": ["The dead do not return."]})
|
|
|
|
bundle = client.get(f"/api/adventures/{client.adv_id}/export").json()
|
|
assert model.owner_of(bundle["narrativeState"], "silver-key") == "mara"
|
|
assert bundle["campaignCanon"]["rules"] == ["The dead do not return."]
|
|
assert any("narrativeStateAfter" in node for node in bundle["actions"])
|
|
|
|
imported = client.post("/api/adventures/import", json=bundle)
|
|
assert imported.status_code == 201, imported.text
|
|
new_id = imported.json()["id"]
|
|
|
|
r = client.get(f"/api/adventures/{new_id}/state")
|
|
assert model.owner_of(r.json()["document"], "silver-key") == "mara"
|
|
|
|
# And the imported history is walkable: undo restores the earlier state
|
|
# from the snapshot that arrived with it.
|
|
assert client.post(f"/api/adventures/{new_id}/undo").status_code == 200
|
|
r = client.get(f"/api/adventures/{new_id}/state")
|
|
assert model.owner_of(r.json()["document"], "silver-key") == "aldric"
|
|
|
|
|
|
def test_a_pre_m5_bundle_imports_with_an_empty_state(client):
|
|
"""Backward compatibility. A file written before M5 has no state section,
|
|
and a campaign that had none is what it records — so it opens, and it opens
|
|
with nothing established rather than with something invented."""
|
|
_play(client, "establish", events=FANTASY)
|
|
bundle = client.get(f"/api/adventures/{client.adv_id}/export").json()
|
|
del bundle["narrativeState"]
|
|
del bundle["campaignCanon"]
|
|
for node in bundle["actions"]:
|
|
node.pop("narrativeStateAfter", None)
|
|
node.pop("stateChanges", None)
|
|
|
|
imported = client.post("/api/adventures/import", json=bundle)
|
|
assert imported.status_code == 201, imported.text
|
|
new_id = imported.json()["id"]
|
|
|
|
r = client.get(f"/api/adventures/{new_id}/state")
|
|
assert r.json()["empty"] is True
|
|
# The story itself arrived intact.
|
|
assert len(client.get(f"/api/adventures/{new_id}").json()["actions"]) > 1
|
|
|
|
|
|
def test_importing_state_does_not_move_the_active_head(client):
|
|
"""The head still comes from the bundle's own `headDepth` (M3/M4)."""
|
|
_play(client, "establish", events=FANTASY)
|
|
_play(client, "second", events=[])
|
|
_play(client, "third", events=[])
|
|
client.post(f"/api/adventures/{client.adv_id}/undo")
|
|
undone = [a["text"] for a in client.get(
|
|
f"/api/adventures/{client.adv_id}").json()["actions"]]
|
|
|
|
imported = client.post(
|
|
"/api/adventures/import",
|
|
json=client.get(f"/api/adventures/{client.adv_id}/export").json())
|
|
new_id = imported.json()["id"]
|
|
|
|
assert [a["text"] for a in client.get(
|
|
f"/api/adventures/{new_id}").json()["actions"]] == undone
|
|
assert client.get(f"/api/adventures/{new_id}").json()["can_redo"] is True
|
|
|
|
|
|
# ================================================================ migration
|
|
|
|
def test_a_pre_m5_database_opens_and_keeps_its_story():
|
|
"""M5 changes a load-bearing persistent subsystem, so a pre-M5 database has
|
|
to survive it: the transcript, the branches, the head, the retries and the
|
|
Save Points all still there, and no narrative state invented for them.
|
|
|
|
The RPG numbers are deliberately **not** translated. `player.gold = 70` says
|
|
nothing about who anyone is, where they stand or what they hold, and a fact
|
|
conjured from it would be fiction the campaign never established.
|
|
"""
|
|
import os
|
|
import tempfile
|
|
from sqlalchemy import create_engine, inspect, text as sql
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app import migrations
|
|
|
|
path = os.path.join(tempfile.mkdtemp(), "m4.db")
|
|
old = create_engine(f"sqlite:///{path}")
|
|
Base.metadata.create_all(bind=old)
|
|
|
|
session = sessionmaker(bind=old)()
|
|
try:
|
|
owner = models.User(is_guest=False, email="m4@example.com")
|
|
session.add(owner)
|
|
session.flush()
|
|
adventure = models.Adventure(
|
|
user_id=owner.id, title="An M4 campaign", head_depth=2,
|
|
world_state={"player": {"gold": 70, "hp": 40}},
|
|
)
|
|
session.add(adventure)
|
|
session.flush()
|
|
branch = models.Branch(adventure_id=adventure.id, lineage=[])
|
|
session.add(branch)
|
|
session.flush()
|
|
branch.lineage = [[branch.id, None]]
|
|
adventure.head_branch_id = branch.id
|
|
for depth in range(3):
|
|
session.add(models.Action(
|
|
adventure_id=adventure.id, branch_id=branch.id, depth=depth,
|
|
type="ai" if depth % 2 else "do", text=f"row {depth}", live=True,
|
|
world_state_after={"player": {"gold": depth * 10}},
|
|
))
|
|
session.add(models.Checkpoint(
|
|
adventure_id=adventure.id, name="Before the abbey",
|
|
branch_id=branch.id, depth=1))
|
|
session.commit()
|
|
adventure_id = adventure.id
|
|
finally:
|
|
session.close()
|
|
|
|
# Make it an M4 *schema*: no M5 tables or columns, stamped at M4's version.
|
|
with old.begin() as conn:
|
|
conn.execute(sql("DROP TABLE state_events"))
|
|
conn.execute(sql("DROP TABLE state_proposals"))
|
|
for column in ("narrative_state", "campaign_canon"):
|
|
conn.execute(sql(f"ALTER TABLE adventures DROP COLUMN {column}"))
|
|
for column in ("narrative_state_after", "state_changes"):
|
|
conn.execute(sql(f"ALTER TABLE actions DROP COLUMN {column}"))
|
|
conn.execute(sql("PRAGMA user_version = 80"))
|
|
|
|
migrations.bootstrap(old)
|
|
|
|
inspector = inspect(old)
|
|
assert "state_events" in inspector.get_table_names()
|
|
assert "state_proposals" in inspector.get_table_names()
|
|
assert "narrative_state" in {c["name"] for c in inspector.get_columns("adventures")}
|
|
assert "narrative_state_after" in {c["name"] for c in inspector.get_columns("actions")}
|
|
|
|
with old.connect() as conn:
|
|
assert conn.execute(sql("PRAGMA user_version")).scalar() == migrations.LATEST_VERSION
|
|
# The story, the head, the branch and the Save Point all survived.
|
|
title, head_depth, head_branch = conn.execute(sql(
|
|
"SELECT title, head_depth, head_branch_id FROM adventures")).one()
|
|
assert (title, head_depth) == ("An M4 campaign", 2)
|
|
assert head_branch is not None
|
|
assert conn.execute(sql("SELECT COUNT(*) FROM actions")).scalar() == 3
|
|
assert conn.execute(sql("SELECT COUNT(*) FROM checkpoints")).scalar() == 1
|
|
# Nothing was invented from the RPG numbers...
|
|
assert conn.execute(sql("SELECT COUNT(*) FROM state_events")).scalar() == 0
|
|
assert conn.execute(sql("SELECT narrative_state FROM adventures")).scalar() is None
|
|
# ...and the legacy values are still there, unread and unharmed.
|
|
assert '"gold": 70' in conn.execute(sql(
|
|
"SELECT world_state FROM adventures")).scalar()
|
|
|
|
# Idempotent.
|
|
migrations.bootstrap(old)
|
|
assert "state_events" in inspect(old).get_table_names()
|
|
old.dispose()
|
|
del adventure_id
|
|
|
|
|
|
# ========================= protocol the block extraction could not remove
|
|
#
|
|
# Both cases below were found by the M5 realistic-context run against a real
|
|
# local model (§12), not imagined. They are the Phase 0B failure class exactly:
|
|
# behaviour that is correct on a small prompt and wrong under a full one.
|
|
|
|
def test_an_echoed_instruction_does_not_reach_the_reader():
|
|
"""A small model reproduces the bracketed reminder it was given, as prose.
|
|
|
|
It arrives with no fence, so nothing that looks for a block strips it, and
|
|
the reader would be shown a piece of the prompt.
|
|
"""
|
|
reply = (
|
|
"You and Mara talk a while longer.\n\n"
|
|
"[Reminder: end your reply with only the ```state block, "
|
|
"without any additional text.]"
|
|
)
|
|
prose, parsed, _raw = extract.split(reply)
|
|
assert prose == "You and Mara talk a while longer."
|
|
assert parsed is None
|
|
|
|
|
|
def test_a_fence_the_model_never_closed_is_not_story():
|
|
"""A model that runs out of output tokens mid-block leaves a dangling
|
|
opener. Everything after it is protocol, so the story ends where it
|
|
begins — otherwise a truncated turn shows the reader half a JSON array."""
|
|
reply = 'The rain falls harder.\n```state\n{"events": [{"type": "add_fact"'
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == "The rain falls harder."
|
|
assert "```" not in prose
|
|
|
|
|
|
# ------------------------------------------- fence safety (review Finding 6)
|
|
#
|
|
# The extractor may remove the application's own protocol payload and nothing
|
|
# else. A story is allowed to contain code, and to talk about the protocol.
|
|
|
|
|
|
def test_a_state_fence_is_always_ours_even_when_it_does_not_parse():
|
|
prose, parsed, raw = extract.split('Beat.\n```state\n{oh no')
|
|
assert prose == "Beat."
|
|
assert parsed is None
|
|
assert raw, "the unparseable payload is kept for the audit"
|
|
|
|
|
|
def test_an_ordinary_json_code_block_stays_in_the_story():
|
|
"""The review's example: a character typing JSON into a terminal kept
|
|
losing their code block, because `json` was treated as our label."""
|
|
reply = ('She typed it out:\n```json\n{"name": "Mara"}\n```\n'
|
|
'Then closed the terminal.')
|
|
prose, parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|
|
assert parsed is None
|
|
|
|
|
|
def test_a_json_fence_that_really_is_a_proposal_is_still_taken():
|
|
"""Models reach for the wrong label; the payload decides, not the word."""
|
|
reply = ('Beat.\n```json\n{"events": [{"type": "add_fact", '
|
|
'"predicate": "the door opened"}]}\n```')
|
|
prose, parsed, _raw = extract.split(reply)
|
|
assert prose == "Beat."
|
|
assert parsed["events"][0]["predicate"] == "the door opened"
|
|
|
|
|
|
def test_a_malformed_proposal_in_a_json_fence_never_reaches_the_reader():
|
|
"""Caught by the realistic-model run during the corrective pass.
|
|
|
|
A small model mangled its own JSON inside a ```json fence. Judging the fence
|
|
only by whether it parsed left the wreckage in the story. A block that
|
|
plainly reads as protocol is ours whether or not it parses — the same rule
|
|
a ```state fence has always had.
|
|
"""
|
|
reply = ('Beat.\n```json\n{"events": [{"type": "set_current_location", '
|
|
'"entity": "you",,}]\n```')
|
|
prose, parsed, raw = extract.split(reply)
|
|
assert prose == "Beat."
|
|
assert '"events"' not in prose
|
|
assert raw, "the malformed payload is still kept for the audit"
|
|
|
|
|
|
def test_a_code_fence_in_another_language_is_never_touched():
|
|
reply = 'He wrote:\n```python\nprint("hi")\n```\nand ran it.'
|
|
prose, parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|
|
assert parsed is None
|
|
|
|
|
|
def test_an_unlabelled_json_shaped_block_that_is_not_a_proposal_stays():
|
|
reply = 'The config read:\n```\n{"name": "Mara"}\n```\nand nothing else.'
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|
|
|
|
|
|
def test_prose_that_merely_mentions_the_protocol_is_kept():
|
|
"""A bracketed aside naming the state block is a sentence in someone's
|
|
story until it looks like the instruction itself."""
|
|
reply = "She frowned. [He was still thinking about the state block.]"
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|
|
|
|
|
|
def test_a_dangling_json_fence_that_is_not_a_proposal_is_kept():
|
|
"""An unterminated code block in a story is still the author's."""
|
|
reply = 'She typed it out:\n```json\n{"name": "Mara"}'
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|
|
|
|
|
|
def test_a_dangling_json_fence_that_is_a_truncated_proposal_is_cut():
|
|
reply = 'Beat.\n```json\n{"events": [{"type": "add_fact"'
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == "Beat."
|
|
|
|
|
|
@pytest.mark.parametrize("reply", [
|
|
'She said, "the vault is sealed" [and it was]',
|
|
"He counted the coins [there were nine] and shrugged.",
|
|
"Plain prose with no protocol at all.",
|
|
])
|
|
def test_ordinary_prose_is_never_trimmed(reply):
|
|
"""The cleanup is narrow on purpose. Removing a sentence from someone's
|
|
story to satisfy a regex is a worse failure than leaving a stray bracket."""
|
|
prose, _parsed, _raw = extract.split(reply)
|
|
assert prose == reply
|