"""M5: turning accepted events into a new state document. Pure and total. Every function here takes a document and returns a new one; none touches the database, and none can fail on an event `validate.review` accepted โ€” validation is the only place an event is refused, so this module never has to decide anything twice. The dispatch is an explicit `if/elif` chain over `events.SPECS`, not a lookup table keyed on the payload. The difference matters: a table maps a string a model supplied to a callable, and the security of that arrangement rests entirely on the allowlist being correct. A chain of literal comparisons cannot be steered by a payload at all, whatever the allowlist does. """ from __future__ import annotations import copy from . import model def apply_events( state: dict, accepted: list[dict], *, branch_id: int | None = None, depth: int | None = None, source: str = "accepted_story", ) -> dict: """Returns `state` with every event in `accepted` applied, in order. The input document is never mutated: head movement stores snapshots by reference in places, and a mutation here would edit the past. `branch_id`/`depth` stamp facts and relationships with where they were established, which is what makes the audit trail answer "which turn caused this" without a join. `source` records whether the campaign, the story or the user established it โ€” C04's provenance, carried on the value itself. """ document = model.normalize(state) for event in accepted: _apply_one(document, event, branch_id, depth, source) return document def _apply_one(state: dict, event: dict, branch_id, depth, source: str) -> None: kind = event["type"] if kind == "create_entity": state["entities"][event["entity"]] = model.new_entity( type=event.get("entity_type") or "other", name=event["name"], description=event.get("description") or "", aliases=event.get("aliases") or [], ) elif kind == "set_entity_status": _entity(state, event["entity"])["status"] = event["status"] elif kind == "set_entity_attribute": # Absolute assignment. The whole reason ADR 010 exists. _entity(state, event["entity"])["attributes"][event["attribute"]] = event["value"] elif kind == "set_entity_conditions": _entity(state, event["entity"])["conditions"] = list(event["conditions"]) elif kind == "set_current_location": _entity(state, event["entity"])["location"] = event["location"] elif kind == "set_possession": state["possessions"][event["item"]] = event["owner"] elif kind == "clear_possession": state["possessions"].pop(event["item"], None) elif kind == "add_fact": state["facts"].append({ "id": event.get("fact_id") or _fact_id(state), "subject": event.get("subject"), "predicate": event["predicate"], "object": event.get("object"), "value": event.get("value"), "authority": _authority(source), "source": source, "status": "active", "branch_id": branch_id, "depth": depth, }) elif kind == "invalidate_fact": for fact in state["facts"]: if fact.get("id") == event["fact_id"]: # Withdrawn, not removed: C04 needs the record of what the # campaign used to believe, and a deleted row audits nothing. fact["status"] = "invalidated" fact["invalidated_by"] = source fact["invalidated_at"] = {"branch_id": branch_id, "depth": depth} if event.get("reason"): fact["invalidated_reason"] = event["reason"] elif kind == "add_relationship": state["relationships"].append({ "id": _relationship_id(state), "source": event["source"], "target": event["target"], "type": event["relationship"], "description": event.get("description") or "", "status": "active", "established_by": source, "branch_id": branch_id, "depth": depth, }) elif kind == "end_relationship": for relationship in state["relationships"]: if ( relationship.get("source") == event["source"] and relationship.get("target") == event["target"] and relationship.get("type") == event["relationship"] and relationship.get("status") == "active" ): relationship["status"] = "ended" relationship["ended_at"] = {"branch_id": branch_id, "depth": depth} elif kind == "open_story_thread": state["threads"][event["thread"]] = { "title": event["title"], "description": event.get("description") or "", "status": "open", "opened_at": {"branch_id": branch_id, "depth": depth}, } elif kind == "resolve_story_thread": thread = state["threads"].get(event["thread"]) if isinstance(thread, dict): thread["status"] = "resolved" thread["resolution"] = event.get("resolution") or "" thread["resolved_at"] = {"branch_id": branch_id, "depth": depth} elif kind == "set_scene": scene = dict(state.get("scene") or {}) if "summary" in event: scene["summary"] = event["summary"] if "location" in event: scene["location"] = event["location"] if "present" in event: scene["present"] = list(event["present"] or []) scene["at"] = {"branch_id": branch_id, "depth": depth} state["scene"] = scene # No `else`. Every allowed type is handled above, and an unhandled one # cannot arrive: `validate.review` refuses anything outside the allowlist, # and the allowlist is this list. A silent fall-through would be the one way # an event could appear accepted and do nothing. def _entity(state: dict, key: str) -> dict: """The entity record for `key`, created bare if a snapshot lost it. Validation guarantees the entity exists, so this is a repair path for a hand-edited or partially imported document rather than a normal branch. A bare record is better than a KeyError: the story is still readable, and the inspector shows an entity with nothing known about it, which is true. """ entities = state["entities"] found = entities.get(key) if not isinstance(found, dict): found = model.new_entity(name=key) entities[key] = found found.setdefault("attributes", {}) found.setdefault("conditions", []) return found def _authority(source: str) -> str: """Which authority band a source's assertions carry. A user's correction outranks the story (C04); the story outranks a guess. `DATA-MODEL.md` ยง14 orders the bands, and this is the mapping into them. """ if source == "manual_correction": return "manual_correction" if source == "campaign_canon": return "campaign_canon" return "accepted_story" def _fact_id(state: dict) -> str: return f"f{len(state['facts']) + 1}" def _relationship_id(state: dict) -> str: return f"r{len(state['relationships']) + 1}" def diff(before: dict, after: dict) -> list[str]: """A short human-readable list of what changed between two documents. Shown under a turn the way the world-state chip used to be, and recorded on the node for the bulk read. Text rather than structure, because its only consumer is a person reading "Aldric now holds the silver key". """ before = model.normalize(before) after = model.normalize(after) lines: list[str] = [] for key, entity in after["entities"].items(): was = before["entities"].get(key) name = model.entity_name(after, key) if was is None: lines.append(f"{name} enters the story") continue if was.get("status") != entity.get("status"): lines.append(f"{name} is now {entity.get('status')}") if was.get("location") != entity.get("location") and entity.get("location"): lines.append(f"{name} is at {model.entity_name(after, entity['location'])}") if sorted(was.get("conditions") or []) != sorted(entity.get("conditions") or []): now = ", ".join(entity.get("conditions") or []) or "nothing" lines.append(f"{name}: {now}") for attribute, value in (entity.get("attributes") or {}).items(): if (was.get("attributes") or {}).get(attribute) != value: lines.append(f"{name} {attribute} = {value}") for item, owner in after["possessions"].items(): if before["possessions"].get(item) != owner: lines.append( f"{model.entity_name(after, item)} โ†’ {model.entity_name(after, owner)}" ) for item in before["possessions"]: if item not in after["possessions"]: lines.append(f"{model.entity_name(after, item)} is held by nobody") known = {f.get("id") for f in before["facts"]} for fact in after["facts"]: if fact.get("id") not in known: lines.append(f"fact: {_fact_text(after, fact)}") was_active = {f["id"] for f in model.active_facts(before)} for fact in before["facts"]: if fact.get("id") in was_active and fact.get("id") not in { f["id"] for f in model.active_facts(after) }: lines.append(f"withdrawn: {_fact_text(after, fact)}") known = {r.get("id") for r in before["relationships"]} for relationship in after["relationships"]: if relationship.get("id") not in known: lines.append( f"{model.entity_name(after, relationship['source'])} " f"{relationship['type']} " f"{model.entity_name(after, relationship['target'])}" ) for key, thread in after["threads"].items(): was = before["threads"].get(key) if was is None: lines.append(f"opened: {thread.get('title', key)}") elif was.get("status") != thread.get("status"): lines.append(f"{thread.get('status')}: {thread.get('title', key)}") return lines def _fact_text(state: dict, fact: dict) -> str: parts = [] if fact.get("subject"): parts.append(model.entity_name(state, fact["subject"])) parts.append(str(fact.get("predicate", ""))) if fact.get("object"): parts.append(model.entity_name(state, fact["object"])) if fact.get("value") is not None: parts.append(str(fact["value"])) return " ".join(p for p in parts if p)