API Reference

The TNGS REST API is served at port 8000. Interactive Swagger UI is available at /docs when the server is running.

Base URL: http://localhost:8000
Authentication: Bearer token (header: Authorization: Bearer <token>)
Content-Type: application/json

---

Health

GET /v1/health/live

Liveness probe. Always returns 200.

Response 200:

{"status": "ok"}

---

GET /v1/health/ready

Readiness probe. Returns 200 when Neo4j is reachable, 503 otherwise.

Response 200:

{"status": "ok", "neo4j": "reachable"}

Response 503:

{"detail": {"status": "degraded", "neo4j": "Connection refused"}}

---

Ingest

POST /v1/notes/import

Ingest raw text, Markdown, or pre-structured JSON. Runs the full pipeline and persists the narrative graph.

Request body:

{
  "title": "My Story",
  "text": "Alice walked slowly. She stopped.\n\nBob arrived at last.",
  "narrative_id": "my-id-001",
  "source_ref": "notebook-2026-04.txt",
  "format": "text"
}

Field	Type	Required	Description
title	string	✓	Narrative title
text	string		Raw prose to ingest
narrative_id	string		Generated if absent
source_ref	string		Provenance reference
format	string		text (default), markdown, json, csv

Response 201:

{
  "narrative_id": "my-id-001",
  "scene_count": 2,
  "atom_count": 3,
  "event_count": 2,
  "character_count": 2,
  "pattern_count": 0,
  "flagged_count": 0
}

---

Narratives

GET /v1/narratives/{id}

Retrieve a narrative's current state.

Path: narrative_id — the narrative's unique ID

Response 200:

{
  "id": "my-id-001",
  "title": "My Story",
  "status": "patterned",
  "source_ref": "notebook-2026-04.txt",
  "scene_count": 2,
  "created_at": "2026-04-26T10:00:00"
}

Response 404: Narrative not found.

---

DELETE /v1/narratives/{id}

Archive a narrative (sets status to archived).

Response 204: No content.
Response 404: Narrative not found.

---

Patterns

POST /v1/patterns

Register a new pattern template.

Request body:

{
  "id": "pattern.pursuit",
  "name": "Pursuit",
  "family": "pursuit",
  "description": "A chase or quest structure."
}

Response 201:

{"id": "pattern.pursuit", "name": "Pursuit", "family": "pursuit", "description": "..."}

---

GET /v1/patterns

List all registered patterns, optionally filtered by family.

Query params: - family (optional) — filter by pattern family string

Response 200: Array of pattern records.

---

GET /v1/patterns/{id}/instances

List concrete pattern instances in a narrative.

Path: pattern_id
Query: narrative_id (required)

Response 200: Array of instance records.
Response 404: Pattern template not found.

---

Transforms

POST /v1/transforms/apply

Apply a transformation axis to a scene.

Request body:

{
  "scene_id": "scene-abc",
  "axis": "pov",
  "parameters": {
    "focalizer": "char-alice",
    "distance": "internal",
    "reliability": "unreliable"
  },
  "operator": "matt"
}

Axis parameter reference:

Axis	Required parameters
pov	focalizer (string)
mood	label (string); optional: valence [-1,1], arousal [0,1]
genre	name (string); optional: conventions (string list)
chronotope	time_mode (cyclical/linear/suspended/compressed), space_mode (bounded/open/liminal/utopian)
reliability	reliability (reliable/unreliable/ambiguous)
code_overlay	atom_id (string), code (hermeneutic/proairetic/semic/symbolic/cultural)

Response 200:

{
  "transform_id": "uuid",
  "scene_id": "scene-abc",
  "axis": "pov",
  "produced_id": "pov-uuid",
  "status": "accepted"
}

Response 400: Invalid axis parameters.
Response 422: Invalid request body schema.

---

GET /v1/transforms/{id}

Retrieve a transform audit record.

Response 200:

{
  "id": "transform-uuid",
  "scene_id": "scene-abc",
  "produced_type": ["Perspective"],
  "produced_id": "pov-uuid"
}

---

Render

POST /v1/render/{id}

Render the current graph state to an output format.

Path: narrative_id

Request body:

{
  "type": "prose",
  "params": {}
}

type value	Output	Content-Type
prose	Markdown prose draft	text/markdown
diff	Transformation diff JSON	application/json
json	Full graph state JSON	application/json
cypher	Replayable MERGE script	text/x-cypher
markdown	Structured summary	text/markdown
graphml	yEd-compatible GraphML with tension-colored edges	application/xml

GraphML / yEd

The graphml render type produces a yEd-compatible GraphML file. Edges are colored on a six-stop gradient (grey → blue → gold → orange → crimson → dark-red) by narrative tension score. See GraphML Export for full details.

Response 200:

{
  "narrative_id": "my-id-001",
  "render_type": "prose",
  "content": "# My Story\n\n## Scene 1\n...",
  "content_type": "text/markdown"
}

Response 404: Narrative not found.
Response 422: Invalid render type.

---

Module Reference

tng.domain.models

Domain model classes for the Transformable Narrative Graph System.

Every class in this module is a pure Pydantic BaseModel. No infrastructure imports (Neo4j driver, FastAPI, etc.) are allowed here. This keeps the domain layer independently testable and decoupled from persistence concerns.

The class hierarchy mirrors the graph schema defined in SRS §4: Narrative → Scene → Atom / Event / PatternInstance Scene → Perspective / MoodState / GenreProfile / Chronotope Atom → CodeTag Transform → (audit trail linking to any scene-level state node)

Atom

Bases: BaseModel

The minimal expressive narrative unit — a single clause or sentence.

Attributes:

Name	Type	Description
id	str	Unique identifier.
text	str	Raw text of the atom.
kind	AtomKind	Functional classification.
surface_order	int	Position within its parent Scene (0-based).
confidence	float	Segmentation / classification confidence in [0.0, 1.0].
code_tags	list[CodeTag]	Barthesian code labels attached to this atom.
needs_review	bool	True when confidence is below the configured threshold.

Source code in src/tng/domain/models.py

class Atom(BaseModel):
    """The minimal expressive narrative unit — a single clause or sentence.

    :ivar id: Unique identifier.
    :ivar text: Raw text of the atom.
    :ivar kind: Functional classification.
    :ivar surface_order: Position within its parent Scene (0-based).
    :ivar confidence: Segmentation / classification confidence in [0.0, 1.0].
    :ivar code_tags: Barthesian code labels attached to this atom.
    :ivar needs_review: True when confidence is below the configured threshold.
    """

    id: str
    text: str
    kind: AtomKind = AtomKind.DESCRIPTIVE
    surface_order: int = 0
    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
    code_tags: list[CodeTag] = Field(default_factory=list)
    needs_review: bool = False

AtomRevision

Bases: BaseModel

A revised version of an Atom's text, preserving the full revision chain.

The graph schema mirrors the transform audit pattern: (Atom)-[:CURRENT_REVISION]->(AtomRevision) points to the latest; (Atom)-[:HAS_REVISION]->(AtomRevision) retains all versions; (AtomRevision)-[:SUPERSEDES]->(AtomRevision) links new → old.

Attributes:

Name	Type	Description
id	str	Unique identifier.
atom_id	str	ID of the parent Atom.
text	str	Revised prose text.
revised_at	datetime	UTC timestamp of the revision.
operator	str	Identifier of the user or system that issued the revision.
reason	str	Optional human-readable reason for the change.

Source code in src/tng/domain/models.py

class AtomRevision(BaseModel):
    """A revised version of an Atom's text, preserving the full revision chain.

    The graph schema mirrors the transform audit pattern:
    ``(Atom)-[:CURRENT_REVISION]->(AtomRevision)`` points to the latest;
    ``(Atom)-[:HAS_REVISION]->(AtomRevision)`` retains all versions;
    ``(AtomRevision)-[:SUPERSEDES]->(AtomRevision)`` links new → old.

    :ivar id: Unique identifier.
    :ivar atom_id: ID of the parent Atom.
    :ivar text: Revised prose text.
    :ivar revised_at: UTC timestamp of the revision.
    :ivar operator: Identifier of the user or system that issued the revision.
    :ivar reason: Optional human-readable reason for the change.
    """

    id: str
    atom_id: str
    text: str
    revised_at: datetime = Field(default_factory=datetime.utcnow)
    operator: str = "system"
    reason: str = ""

Character

Bases: BaseModel

A participant or focalizer in the narrative.

Attributes:

Name	Type	Description
id	str	Unique identifier.
name	str	Character name as it appears in the source text.
role	str	Narrative role (e.g. "protagonist", "antagonist", "witness").

Source code in src/tng/domain/models.py

class Character(BaseModel):
    """A participant or focalizer in the narrative.

    :ivar id: Unique identifier.
    :ivar name: Character name as it appears in the source text.
    :ivar role: Narrative role (e.g. "protagonist", "antagonist", "witness").
    """

    model_config = ConfigDict(frozen=True)

    id: str
    name: str
    role: str = "character"

Chronotope

Bases: BaseModel

Bakhtinian time-space frame for a Scene.

Attributes:

Name	Type	Description
id	str	Unique identifier.
time_mode	str	One of: cyclical, linear, suspended, compressed.
space_mode	str	One of: bounded, open, liminal, utopian.

Source code in src/tng/domain/models.py

class Chronotope(BaseModel):
    """Bakhtinian time-space frame for a Scene.

    :ivar id: Unique identifier.
    :ivar time_mode: One of: cyclical, linear, suspended, compressed.
    :ivar space_mode: One of: bounded, open, liminal, utopian.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    time_mode: str
    space_mode: str

CodeTag

Bases: BaseModel

A Barthesian code label attached to an Atom.

Attributes:

Name	Type	Description
id	str	Unique identifier for this tag.
code	BarthesCode	The Barthesian code category.
label	str	Human-readable annotation label.

Source code in src/tng/domain/models.py

class CodeTag(BaseModel):
    """A Barthesian code label attached to an Atom.

    :ivar id: Unique identifier for this tag.
    :ivar code: The Barthesian code category.
    :ivar label: Human-readable annotation label.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    code: BarthesCode
    label: str

Event

Bases: BaseModel

An action-bearing narrative unit extracted from an Atom.

Attributes:

Name	Type	Description
id	str	Unique identifier.
verb	str	Lemmatised main verb of the event clause.
tense	str	Grammatical tense string (e.g. "past", "present").
aspect	str	Grammatical aspect string (e.g. "simple", "progressive").
confidence	float	Extraction confidence in [0.0, 1.0].
participants	list[Character]	Characters who take part in this event.
needs_review	bool	True when confidence is below the configured threshold.

Source code in src/tng/domain/models.py

class Event(BaseModel):
    """An action-bearing narrative unit extracted from an Atom.

    :ivar id: Unique identifier.
    :ivar verb: Lemmatised main verb of the event clause.
    :ivar tense: Grammatical tense string (e.g. "past", "present").
    :ivar aspect: Grammatical aspect string (e.g. "simple", "progressive").
    :ivar confidence: Extraction confidence in [0.0, 1.0].
    :ivar participants: Characters who take part in this event.
    :ivar needs_review: True when confidence is below the configured threshold.
    """

    id: str
    verb: str
    tense: str = "past"
    aspect: str = "simple"
    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
    participants: list[Character] = Field(default_factory=list)
    needs_review: bool = False

EventRelation

Bases: BaseModel

A directed relationship between two Event nodes.

Captures explicit causal and temporal connections that are stored as first-class relationships in the graph. These are fetched separately from the event nodes themselves because they are inter-event edges rather than containment relationships.

Attributes:

Name	Type	Description
source_id	str	ID of the originating Event.
target_id	str	ID of the destination Event.
relation_type	str	One of CAUSES, ENABLES, PREVENTS, PRECEDES.

Source code in src/tng/domain/models.py

class EventRelation(BaseModel):
    """A directed relationship between two Event nodes.

    Captures explicit causal and temporal connections that are stored as
    first-class relationships in the graph.  These are fetched separately
    from the event nodes themselves because they are inter-event edges
    rather than containment relationships.

    :ivar source_id: ID of the originating Event.
    :ivar target_id: ID of the destination Event.
    :ivar relation_type: One of ``CAUSES``, ``ENABLES``, ``PREVENTS``,
        ``PRECEDES``.
    """

    model_config = ConfigDict(frozen=True)

    source_id: str
    target_id: str
    relation_type: str

GenreProfile

Bases: BaseModel

Genre encoding for a Scene or Narrative.

Attributes:

Name	Type	Description
id	str	Unique identifier.
name	str	Genre name (e.g. "gothic", "road novel").
conventions	list[str]	JSON-serialisable list of constraint strings describing genre-specific narrative obligations.

Source code in src/tng/domain/models.py

class GenreProfile(BaseModel):
    """Genre encoding for a Scene or Narrative.

    :ivar id: Unique identifier.
    :ivar name: Genre name (e.g. "gothic", "road novel").
    :ivar conventions: JSON-serialisable list of constraint strings describing
        genre-specific narrative obligations.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    name: str
    conventions: list[str] = Field(default_factory=list)

GraphState

Bases: BaseModel

A complete, self-contained snapshot of one narrative's graph state.

Passed to renderer implementations so they never issue Cypher directly.

Attributes:

Name	Type	Description
narrative	Narrative	The root Narrative with all nested scenes and atoms.
transforms	list[Transform]	Ordered transform history (oldest first).
characters	list[Character]	All Characters referenced in this narrative.
event_relations	list[EventRelation]	Explicit inter-event relationships (CAUSES, ENABLES, PREVENTS, PRECEDES) fetched from the graph. Used by the GraphML renderer to draw and score causal/temporal edges.

Source code in src/tng/domain/models.py

class GraphState(BaseModel):
    """A complete, self-contained snapshot of one narrative's graph state.

    Passed to renderer implementations so they never issue Cypher directly.

    :ivar narrative: The root Narrative with all nested scenes and atoms.
    :ivar transforms: Ordered transform history (oldest first).
    :ivar characters: All Characters referenced in this narrative.
    :ivar event_relations: Explicit inter-event relationships (CAUSES,
        ENABLES, PREVENTS, PRECEDES) fetched from the graph.  Used by the
        GraphML renderer to draw and score causal/temporal edges.
    """

    narrative: Narrative
    transforms: list[Transform] = Field(default_factory=list)
    characters: list[Character] = Field(default_factory=list)
    event_relations: list[EventRelation] = Field(default_factory=list)

MoodState

Bases: BaseModel

Affective/tonal state for a Scene.

Attributes:

Name	Type	Description
id	str	Unique identifier.
label	str	Free-text mood label (e.g. "melancholic", "tense").
valence	float	Sentiment polarity in [-1.0, 1.0]; negative = negative affect.
arousal	float	Activation level in [0.0, 1.0]; high = energetic.

Source code in src/tng/domain/models.py

class MoodState(BaseModel):
    """Affective/tonal state for a Scene.

    :ivar id: Unique identifier.
    :ivar label: Free-text mood label (e.g. "melancholic", "tense").
    :ivar valence: Sentiment polarity in [-1.0, 1.0]; negative = negative affect.
    :ivar arousal: Activation level in [0.0, 1.0]; high = energetic.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    label: str
    valence: float = Field(default=0.0, ge=-1.0, le=1.0)
    arousal: float = Field(default=0.5, ge=0.0, le=1.0)

Narrative

Bases: BaseModel

Top-level work or draft — the root node of a TNGS narrative graph.

Attributes:

Name	Type	Description
id	str	Unique identifier.
title	str	Working title of the narrative.
status	NarrativeStatus	Life-cycle state.
source_ref	str	Optional reference to the originating source document.
scenes	list[Scene]	Ordered list of Scenes.
created_at	datetime	UTC creation timestamp.

Source code in src/tng/domain/models.py

class Narrative(BaseModel):
    """Top-level work or draft — the root node of a TNGS narrative graph.

    :ivar id: Unique identifier.
    :ivar title: Working title of the narrative.
    :ivar status: Life-cycle state.
    :ivar source_ref: Optional reference to the originating source document.
    :ivar scenes: Ordered list of Scenes.
    :ivar created_at: UTC creation timestamp.
    """

    id: str
    title: str
    status: NarrativeStatus = NarrativeStatus.DRAFT
    source_ref: str = ""
    scenes: list[Scene] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.utcnow)

Pattern

Bases: BaseModel

A reusable narrative template stored in the graph library.

Attributes:

Name	Type	Description
id	str	Unique identifier (e.g. "pattern.gift_exchange").
name	str	Human-readable name.
family	str	Family tag (see PatternFamily enum).
description	str	Prose description of the pattern's narrative function.

Source code in src/tng/domain/models.py

class Pattern(BaseModel):
    """A reusable narrative template stored in the graph library.

    :ivar id: Unique identifier (e.g. "pattern.gift_exchange").
    :ivar name: Human-readable name.
    :ivar family: Family tag (see PatternFamily enum).
    :ivar description: Prose description of the pattern's narrative function.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    name: str
    family: str
    description: str = ""

PatternInstance

Bases: BaseModel

Concrete realisation of a Pattern in a specific Scene.

Attributes:

Name	Type	Description
id	str	Unique identifier.
slot	str	Structural slot label (e.g. "scene-core", "opening").
confidence	float	Match confidence in [0.0, 1.0].
template	Pattern | None	The Pattern this instance realises.
realized_atoms		Atom IDs that ground this instance.
realized_events		Event IDs that ground this instance.
needs_review	bool	True when confidence is below the configured threshold.

Source code in src/tng/domain/models.py

class PatternInstance(BaseModel):
    """Concrete realisation of a Pattern in a specific Scene.

    :ivar id: Unique identifier.
    :ivar slot: Structural slot label (e.g. "scene-core", "opening").
    :ivar confidence: Match confidence in [0.0, 1.0].
    :ivar template: The Pattern this instance realises.
    :ivar realized_atoms: Atom IDs that ground this instance.
    :ivar realized_events: Event IDs that ground this instance.
    :ivar needs_review: True when confidence is below the configured threshold.
    """

    id: str
    slot: str = "scene-core"
    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
    template: Pattern | None = None
    realized_atom_ids: list[str] = Field(default_factory=list)
    realized_event_ids: list[str] = Field(default_factory=list)
    needs_review: bool = False

Perspective

Bases: BaseModel

Focalization state for a Scene at a point in transformation history.

Attributes:

Name	Type	Description
id	str	Unique identifier.
focalizer	str	ID of the Character through whose perspective events are filtered.
distance	FocalizationDistance	Genettean focalization distance.
reliability	ReliabilityLevel	Narrator/focalizer credibility rating.

Source code in src/tng/domain/models.py

class Perspective(BaseModel):
    """Focalization state for a Scene at a point in transformation history.

    :ivar id: Unique identifier.
    :ivar focalizer: ID of the Character through whose perspective events are filtered.
    :ivar distance: Genettean focalization distance.
    :ivar reliability: Narrator/focalizer credibility rating.
    """

    model_config = ConfigDict(frozen=True)

    id: str
    focalizer: str
    distance: FocalizationDistance = FocalizationDistance.ZERO
    reliability: ReliabilityLevel = ReliabilityLevel.RELIABLE

Scene

Bases: BaseModel

A bounded narrative segment within a Narrative.

Attributes:

Name	Type	Description
id	str	Unique identifier.
sequence	int	Ordinal position within the parent Narrative (1-based).
summary	str	Optional human-readable summary of the scene.
atoms	list[Atom]	Ordered list of Atoms in this scene.
events	list[Event]	Events extracted from this scene.
pattern_instances	list[PatternInstance]	Pattern instances detected in this scene.
current_perspective	Perspective | None	Active Perspective node (if any).
current_mood	MoodState | None	Active MoodState node (if any).
current_genre	GenreProfile | None	Active GenreProfile node (if any).
chronotope	Chronotope | None	Active Chronotope node (if any).

Source code in src/tng/domain/models.py

class Scene(BaseModel):
    """A bounded narrative segment within a Narrative.

    :ivar id: Unique identifier.
    :ivar sequence: Ordinal position within the parent Narrative (1-based).
    :ivar summary: Optional human-readable summary of the scene.
    :ivar atoms: Ordered list of Atoms in this scene.
    :ivar events: Events extracted from this scene.
    :ivar pattern_instances: Pattern instances detected in this scene.
    :ivar current_perspective: Active Perspective node (if any).
    :ivar current_mood: Active MoodState node (if any).
    :ivar current_genre: Active GenreProfile node (if any).
    :ivar chronotope: Active Chronotope node (if any).
    """

    id: str
    sequence: int = 1
    summary: str = ""
    atoms: list[Atom] = Field(default_factory=list)
    events: list[Event] = Field(default_factory=list)
    pattern_instances: list[PatternInstance] = Field(default_factory=list)
    current_perspective: Perspective | None = None
    current_mood: MoodState | None = None
    current_genre: GenreProfile | None = None
    chronotope: Chronotope | None = None

Transform

Bases: BaseModel

Audit record for a single transformation operation.

The Transform node is the spine of the transformation lineage graph. It links the scene it modified (APPLIED_TO) and the new state node it produced (PRODUCED). It is never deleted or overwritten; the full sequence of transforms is always traversable.

Attributes:

Name	Type	Description
id	str	Unique identifier.
axis	TransformAxis	The transformation axis that was applied.
operator	str	Identifier of the user or system that issued the transform.
applied_at	datetime	UTC timestamp of the operation.
parameters	dict[str, Any]	Axis-specific parameters as a free dict (serialised to JSON when persisted).
scene_id	str	ID of the scene this transform was applied to.
produced_id	str	ID of the new state node produced by this transform.

Source code in src/tng/domain/models.py

class Transform(BaseModel):
    """Audit record for a single transformation operation.

    The Transform node is the spine of the transformation lineage graph.
    It links the scene it modified (``APPLIED_TO``) and the new state node
    it produced (``PRODUCED``).  It is never deleted or overwritten; the
    full sequence of transforms is always traversable.

    :ivar id: Unique identifier.
    :ivar axis: The transformation axis that was applied.
    :ivar operator: Identifier of the user or system that issued the transform.
    :ivar applied_at: UTC timestamp of the operation.
    :ivar parameters: Axis-specific parameters as a free dict (serialised to
        JSON when persisted).
    :ivar scene_id: ID of the scene this transform was applied to.
    :ivar produced_id: ID of the new state node produced by this transform.
    """

    id: str
    axis: TransformAxis
    operator: str = "system"
    applied_at: datetime = Field(default_factory=datetime.utcnow)
    parameters: dict[str, Any] = Field(default_factory=dict)
    scene_id: str = ""
    produced_id: str = ""

tng.domain.enums

Bounded enumeration types for the TNGS domain vocabulary.

All semantic axes that must stay closed (cannot be arbitrary strings) are defined here. Using enumerations rather than open strings makes every vocabulary set queryable by value, prevents typo-driven divergence, and guarantees that Pydantic validation catches unknown values at the API boundary rather than silently persisting garbage to the graph.

AtomKind

Bases: str, Enum

Functional classification of a narrative atom.

Attributes:

Name	Type	Description
DESCRIPTIVE		Depicts setting, appearance, or state.
DIALOGIC		Direct or indirect speech act.
REFLEXIVE		Character introspection or narratorial comment.
TRANSITIONAL		Moves the narrative between scenes or moments.
EXPOSITORY		Background information, world-building, or backstory.

Source code in src/tng/domain/enums.py

class AtomKind(str, Enum):
    """Functional classification of a narrative atom.

    :cvar DESCRIPTIVE: Depicts setting, appearance, or state.
    :cvar DIALOGIC: Direct or indirect speech act.
    :cvar REFLEXIVE: Character introspection or narratorial comment.
    :cvar TRANSITIONAL: Moves the narrative between scenes or moments.
    :cvar EXPOSITORY: Background information, world-building, or backstory.
    """

    DESCRIPTIVE = "descriptive"
    DIALOGIC = "dialogic"
    REFLEXIVE = "reflexive"
    TRANSITIONAL = "transitional"
    EXPOSITORY = "expository"

BarthesCode

Bases: str, Enum

Barthesian narrative codes (from S/Z, Roland Barthes, 1970).

Attributes:

Name	Type	Description
HERMENEUTIC		Mystery or enigma that propels reader anticipation.
PROAIRETIC		Action sequence implying a consequent action.
SEMIC		Connotative detail that builds character or atmosphere.
SYMBOLIC		Binary or antithetical thematic opposition.
CULTURAL		Reference to a shared body of knowledge or convention.

Source code in src/tng/domain/enums.py

class BarthesCode(str, Enum):
    """Barthesian narrative codes (from *S/Z*, Roland Barthes, 1970).

    :cvar HERMENEUTIC: Mystery or enigma that propels reader anticipation.
    :cvar PROAIRETIC: Action sequence implying a consequent action.
    :cvar SEMIC: Connotative detail that builds character or atmosphere.
    :cvar SYMBOLIC: Binary or antithetical thematic opposition.
    :cvar CULTURAL: Reference to a shared body of knowledge or convention.
    """

    HERMENEUTIC = "hermeneutic"
    PROAIRETIC = "proairetic"
    SEMIC = "semic"
    SYMBOLIC = "symbolic"
    CULTURAL = "cultural"

FocalizationDistance

Bases: str, Enum

Genettean focalization distance for a Perspective node.

Attributes:

Name	Type	Description
ZERO		Narrator knows more than any character (omniscient).
INTERNAL		Narrative filtered through one character's consciousness.
EXTERNAL		Narrator records behaviour without access to interiority.

Source code in src/tng/domain/enums.py

class FocalizationDistance(str, Enum):
    """Genettean focalization distance for a Perspective node.

    :cvar ZERO: Narrator knows more than any character (omniscient).
    :cvar INTERNAL: Narrative filtered through one character's consciousness.
    :cvar EXTERNAL: Narrator records behaviour without access to interiority.
    """

    ZERO = "zero"
    INTERNAL = "internal"
    EXTERNAL = "external"

NarrativeStatus

Bases: str, Enum

Life-cycle states for a Narrative node (see state machine, SRS §7.4).

Attributes:

Name	Type	Description
DRAFT		Narrative created but not yet atomized.
ATOMIZED		Ingest complete; atoms and events persisted.
PATTERNED		Pattern detection run; instances linked.
TRANSFORMED		At least one transformation axis applied.
RENDERED		Render operation has been performed.
EXPORTED		Narrative exported to an external format.
ARCHIVED		Administratively archived; no further processing.

Source code in src/tng/domain/enums.py

class NarrativeStatus(str, Enum):
    """Life-cycle states for a Narrative node (see state machine, SRS §7.4).

    :cvar DRAFT: Narrative created but not yet atomized.
    :cvar ATOMIZED: Ingest complete; atoms and events persisted.
    :cvar PATTERNED: Pattern detection run; instances linked.
    :cvar TRANSFORMED: At least one transformation axis applied.
    :cvar RENDERED: Render operation has been performed.
    :cvar EXPORTED: Narrative exported to an external format.
    :cvar ARCHIVED: Administratively archived; no further processing.
    """

    DRAFT = "draft"
    ATOMIZED = "atomized"
    PATTERNED = "patterned"
    TRANSFORMED = "transformed"
    RENDERED = "rendered"
    EXPORTED = "exported"
    ARCHIVED = "archived"

PatternFamily

Bases: str, Enum

High-level family tags for narrative pattern templates.

Attributes:

Name	Type	Description
RITUAL		Socially codified exchange or ceremony.
TRANSITION		Movement across a threshold or boundary.
CONFLICT		Antagonistic encounter between agents.
REVELATION		Disclosure of previously hidden information.
PURSUIT		Chase or quest structure.
TRANSFORMATION		Internal or external change of state.

Source code in src/tng/domain/enums.py

class PatternFamily(str, Enum):
    """High-level family tags for narrative pattern templates.

    :cvar RITUAL: Socially codified exchange or ceremony.
    :cvar TRANSITION: Movement across a threshold or boundary.
    :cvar CONFLICT: Antagonistic encounter between agents.
    :cvar REVELATION: Disclosure of previously hidden information.
    :cvar PURSUIT: Chase or quest structure.
    :cvar TRANSFORMATION: Internal or external change of state.
    """

    RITUAL = "ritual"
    TRANSITION = "transition"
    CONFLICT = "conflict"
    REVELATION = "revelation"
    PURSUIT = "pursuit"
    TRANSFORMATION = "transformation"

ReliabilityLevel

Bases: str, Enum

Credibility rating assigned to a narrative Perspective.

Attributes:

Name	Type	Description
RELIABLE		Narrator or focalizer is trustworthy.
UNRELIABLE		Narrator or focalizer is demonstrably biased or wrong.
AMBIGUOUS		Reliability cannot be determined from available evidence.

Source code in src/tng/domain/enums.py

class ReliabilityLevel(str, Enum):
    """Credibility rating assigned to a narrative Perspective.

    :cvar RELIABLE: Narrator or focalizer is trustworthy.
    :cvar UNRELIABLE: Narrator or focalizer is demonstrably biased or wrong.
    :cvar AMBIGUOUS: Reliability cannot be determined from available evidence.
    """

    RELIABLE = "reliable"
    UNRELIABLE = "unreliable"
    AMBIGUOUS = "ambiguous"

RenderType

Bases: str, Enum

Output format requested from the Render endpoint.

Attributes:

Name	Type	Description
PROSE		Atoms in surface order as a prose draft.
DIFF		Side-by-side before/after for each transformed axis.
JSON		Full graph state as a JSON document.
CYPHER		Replayable Cypher MERGE script.
MARKDOWN		Structured Markdown summary with transform log.
GRAPHML		yEd-compatible GraphML with edges coloured by narrative tension. Suitable for visual graph exploration in yEd or any tool that reads the GraphML + yFiles extension schema.

Source code in src/tng/domain/enums.py

class RenderType(str, Enum):
    """Output format requested from the Render endpoint.

    :cvar PROSE: Atoms in surface order as a prose draft.
    :cvar DIFF: Side-by-side before/after for each transformed axis.
    :cvar JSON: Full graph state as a JSON document.
    :cvar CYPHER: Replayable Cypher MERGE script.
    :cvar MARKDOWN: Structured Markdown summary with transform log.
    :cvar GRAPHML: yEd-compatible GraphML with edges coloured by narrative
        tension.  Suitable for visual graph exploration in yEd or any tool
        that reads the GraphML + yFiles extension schema.
    """

    PROSE = "prose"
    DIFF = "diff"
    JSON = "json"
    CYPHER = "cypher"
    MARKDOWN = "markdown"
    GRAPHML = "graphml"

TransformAxis

Bases: str, Enum

The six supported transformation axes of TNGS.

Each axis operates on a specific domain object and produces a new state node rather than overwriting the existing one, preserving the full transformation lineage as traversable graph state.

Attributes:

Name	Type	Description
POV		Point-of-view / focalization shift.
MOOD		Affective / tonal retag.
GENRE		Genre profile swap.
CHRONOTOPE		Bakhtinian time-space remap.
RELIABILITY		Narrator reliability adjustment.
CODE_OVERLAY		Barthesian code attachment to atoms.

Source code in src/tng/domain/enums.py

class TransformAxis(str, Enum):
    """The six supported transformation axes of TNGS.

    Each axis operates on a specific domain object and produces a new
    state node rather than overwriting the existing one, preserving the
    full transformation lineage as traversable graph state.

    :cvar POV: Point-of-view / focalization shift.
    :cvar MOOD: Affective / tonal retag.
    :cvar GENRE: Genre profile swap.
    :cvar CHRONOTOPE: Bakhtinian time-space remap.
    :cvar RELIABILITY: Narrator reliability adjustment.
    :cvar CODE_OVERLAY: Barthesian code attachment to atoms.
    """

    POV = "pov"
    MOOD = "mood"
    GENRE = "genre"
    CHRONOTOPE = "chronotope"
    RELIABILITY = "reliability"
    CODE_OVERLAY = "code_overlay"

tng.services.ingest_service

Ingest service — orchestrates the full ingest pipeline (SRS §6.1, Diagram 7).

Responsibilities

1. Accept a raw text payload (plain text, Markdown, or pre-structured JSON).
2. Segment text into scenes and atoms via the segmenter.
3. Extract entities and events via entity_extractor and event_detector.
4. Apply confidence scoring and review flags via annotator.
5. Delegate pattern detection to PatternService.
6. Persist the complete result via GraphRepository in a single pass.
7. Return an IngestResult summary to the caller.

The service never issues Cypher directly; it only calls the repository. All pre-processing runs in memory before any graph write, ensuring the graph is never left in a partially-atomized state (SRS Diagram 7 note).

IngestPayload dataclass

Normalised input payload for the IngestService.

Parameters:

Name	Type	Description	Default
title	str	Narrative title.	required
text	str	Raw prose or pre-structured text to ingest.	required
narrative_id	str	Optional; generated if absent.	make_id()
source_ref	str	Optional provenance reference.	''
format	str	Input format hint: "text", "markdown", or "json".	'text'
annotations	dict	Optional pre-annotations dict (from JSON payloads).	dict()

Source code in src/tng/services/ingest_service.py

@dataclass
class IngestPayload:
    """Normalised input payload for the IngestService.

    :param title: Narrative title.
    :param text: Raw prose or pre-structured text to ingest.
    :param narrative_id: Optional; generated if absent.
    :param source_ref: Optional provenance reference.
    :param format: Input format hint: ``"text"``, ``"markdown"``, or ``"json"``.
    :param annotations: Optional pre-annotations dict (from JSON payloads).
    """

    title: str
    text: str
    narrative_id: str = field(default_factory=make_id)
    source_ref: str = ""
    format: str = "text"
    annotations: dict = field(default_factory=dict)

IngestResult dataclass

Summary returned to the API after a successful ingest operation.

Parameters:

Name	Type	Description	Default
narrative_id	str	ID of the created or updated Narrative.	required
scene_count	int	Number of scenes persisted.	0
atom_count	int	Total atoms written to the graph.	0
event_count	int	Total events written.	0
character_count	int	Total characters written.	0
pattern_count	int	Number of pattern instances created.	0
flagged_count	int	Number of nodes flagged for human review.	0

Source code in src/tng/services/ingest_service.py

@dataclass
class IngestResult:
    """Summary returned to the API after a successful ingest operation.

    :param narrative_id: ID of the created or updated Narrative.
    :param scene_count: Number of scenes persisted.
    :param atom_count: Total atoms written to the graph.
    :param event_count: Total events written.
    :param character_count: Total characters written.
    :param pattern_count: Number of pattern instances created.
    :param flagged_count: Number of nodes flagged for human review.
    """

    narrative_id: str
    scene_count: int = 0
    atom_count: int = 0
    event_count: int = 0
    character_count: int = 0
    pattern_count: int = 0
    flagged_count: int = 0

IngestService

Orchestrates the ingest pipeline from raw text to persisted graph.

Parameters:

Name	Type	Description	Default
repo	GraphRepository	An open GraphRepository instance.	required
pattern_service	PatternService	A PatternService for pattern detection.	required
settings	Settings	Application settings (used for confidence_threshold).	required

Source code in src/tng/services/ingest_service.py

class IngestService:
    """Orchestrates the ingest pipeline from raw text to persisted graph.

    :param repo: An open ``GraphRepository`` instance.
    :param pattern_service: A ``PatternService`` for pattern detection.
    :param settings: Application settings (used for ``confidence_threshold``).
    """

    def __init__(
        self,
        repo: GraphRepository,
        pattern_service: PatternService,
        settings: Settings,
    ) -> None:
        self._repo = repo
        self._pattern_service = pattern_service
        self._threshold = settings.confidence_threshold

    def ingest(self, payload: IngestPayload) -> IngestResult:
        """Run the full ingest pipeline and persist the result.

        :param payload: Normalised input payload.
        :returns: ``IngestResult`` summary.
        """
        logger.info("Starting ingest for narrative %r", payload.narrative_id)

        sections = self._segment(payload)

        narrative = Narrative(
            id=payload.narrative_id,
            title=payload.title,
            status=NarrativeStatus.DRAFT,
            source_ref=payload.source_ref,
        )
        self._repo.save_narrative(narrative)

        total_atoms = total_events = total_chars = total_patterns = flagged = 0

        for seq, section in enumerate(sections, start=1):
            atoms = annotate_atoms(section.sentences, self._threshold)
            entities = extract_entities(section.sentences, self._threshold)
            characters = annotate_characters(entities)
            detected = detect_events(section.sentences, self._threshold)
            events = annotate_events(detected, characters)

            pattern_instances = self._pattern_service.detect_patterns(
                atoms, events, payload.narrative_id
            )

            scene = Scene(
                id=make_id(),
                sequence=seq,
                summary=section.summary,
                atoms=atoms,
                events=events,
                pattern_instances=pattern_instances,
            )

            self._repo.save_scene(scene, payload.narrative_id)

            total_atoms += len(atoms)
            total_events += len(events)
            total_chars += len(characters)
            total_patterns += len(pattern_instances)
            flagged += sum(1 for a in atoms if a.needs_review)
            flagged += sum(1 for e in events if e.needs_review)

        self._repo.update_narrative_status(
            payload.narrative_id, NarrativeStatus.PATTERNED
        )
        logger.info(
            "Ingest complete: atoms=%d events=%d patterns=%d flagged=%d",
            total_atoms,
            total_events,
            total_patterns,
            flagged,
        )
        return IngestResult(
            narrative_id=payload.narrative_id,
            scene_count=len(sections),
            atom_count=total_atoms,
            event_count=total_events,
            character_count=total_chars,
            pattern_count=total_patterns,
            flagged_count=flagged,
        )

    def _segment(self, payload: IngestPayload) -> list[SceneSection]:
        """Segment the payload text into scene sections.

        Dispatches to ``segment_markdown`` when ``format == "markdown"`` so
        heading lines become scene boundaries with a populated ``summary``.
        All other formats use the paragraph-boundary segmenter and produce
        sections with ``summary = ""``.

        :param payload: The ingest payload.
        :returns: Ordered list of ``SceneSection`` instances.
        """
        text = payload.text.strip()
        if payload.format == "markdown":
            return segment_markdown(text)
        segmented = segment_text(strip_markdown_frontmatter(text))
        return [
            SceneSection(summary="", sentences=sentences)
            for sentences in segmented.sentences_by_paragraph
        ]

ingest(payload)

Run the full ingest pipeline and persist the result.

Parameters:

Name	Type	Description	Default
payload	IngestPayload	Normalised input payload.	required

Returns:

Type	Description
IngestResult	IngestResult summary.

Source code in src/tng/services/ingest_service.py

def ingest(self, payload: IngestPayload) -> IngestResult:
    """Run the full ingest pipeline and persist the result.

    :param payload: Normalised input payload.
    :returns: ``IngestResult`` summary.
    """
    logger.info("Starting ingest for narrative %r", payload.narrative_id)

    sections = self._segment(payload)

    narrative = Narrative(
        id=payload.narrative_id,
        title=payload.title,
        status=NarrativeStatus.DRAFT,
        source_ref=payload.source_ref,
    )
    self._repo.save_narrative(narrative)

    total_atoms = total_events = total_chars = total_patterns = flagged = 0

    for seq, section in enumerate(sections, start=1):
        atoms = annotate_atoms(section.sentences, self._threshold)
        entities = extract_entities(section.sentences, self._threshold)
        characters = annotate_characters(entities)
        detected = detect_events(section.sentences, self._threshold)
        events = annotate_events(detected, characters)

        pattern_instances = self._pattern_service.detect_patterns(
            atoms, events, payload.narrative_id
        )

        scene = Scene(
            id=make_id(),
            sequence=seq,
            summary=section.summary,
            atoms=atoms,
            events=events,
            pattern_instances=pattern_instances,
        )

        self._repo.save_scene(scene, payload.narrative_id)

        total_atoms += len(atoms)
        total_events += len(events)
        total_chars += len(characters)
        total_patterns += len(pattern_instances)
        flagged += sum(1 for a in atoms if a.needs_review)
        flagged += sum(1 for e in events if e.needs_review)

    self._repo.update_narrative_status(
        payload.narrative_id, NarrativeStatus.PATTERNED
    )
    logger.info(
        "Ingest complete: atoms=%d events=%d patterns=%d flagged=%d",
        total_atoms,
        total_events,
        total_patterns,
        flagged,
    )
    return IngestResult(
        narrative_id=payload.narrative_id,
        scene_count=len(sections),
        atom_count=total_atoms,
        event_count=total_events,
        character_count=total_chars,
        pattern_count=total_patterns,
        flagged_count=flagged,
    )

tng.services.transform_service

Transform service — validates and applies transformation axes (SRS §7).

The six transformation axes are each dispatched through a dedicated validator and then routed to the GraphRepository. The service is the only place where axis-specific parameter validation occurs; the repository is responsible only for the Cypher mechanics.

Non-destructive contract (SRS §7.2)

Every transformation:

1. Creates a new state node.
2. Detaches the old CURRENT_* relationship.
3. Attaches the new CURRENT_* relationship.
4. Creates a Transform audit node linked to both.

The old state node is never deleted — full lineage is always traversable.

Design notes

• Axis validators use Pydantic for parameter schemas; an invalid parameter dict raises ValueError before any graph write occurs (SRS Diagram 9).
• The TransformRequest dataclass is the public input contract; it is populated from the API schema and handed to apply().

ChronotopeParams

Bases: BaseModel

Parameters for a chronotope transformation.

Attributes:

Name	Type	Description
time_mode	str	Time mode (cyclical/linear/suspended/compressed).
space_mode	str	Space mode (bounded/open/liminal/utopian).

Source code in src/tng/services/transform_service.py

class ChronotopeParams(BaseModel):
    """Parameters for a chronotope transformation.

    :ivar time_mode: Time mode (cyclical/linear/suspended/compressed).
    :ivar space_mode: Space mode (bounded/open/liminal/utopian).
    """

    time_mode: str
    space_mode: str

    @field_validator("time_mode")
    @classmethod
    def _validate_time_mode(cls, v: str) -> str:
        valid = {"cyclical", "linear", "suspended", "compressed"}
        if v not in valid:
            raise ValueError(f"time_mode must be one of {valid}")
        return v

    @field_validator("space_mode")
    @classmethod
    def _validate_space_mode(cls, v: str) -> str:
        valid = {"bounded", "open", "liminal", "utopian"}
        if v not in valid:
            raise ValueError(f"space_mode must be one of {valid}")
        return v

CodeOverlayParams

Bases: BaseModel

Parameters for a code overlay transformation.

Attributes:

Name	Type	Description
atom_id	str	ID of the target Atom.
code	BarthesCode	Barthesian code category.
label	str	Human-readable annotation label.

Source code in src/tng/services/transform_service.py

class CodeOverlayParams(BaseModel):
    """Parameters for a code overlay transformation.

    :ivar atom_id: ID of the target Atom.
    :ivar code: Barthesian code category.
    :ivar label: Human-readable annotation label.
    """

    atom_id: str
    code: BarthesCode
    label: str = ""

GenreParams

Bases: BaseModel

Parameters for a genre transformation.

Attributes:

Name	Type	Description
name	str	Genre name.
conventions	list[str]	List of genre constraint strings.

Source code in src/tng/services/transform_service.py

class GenreParams(BaseModel):
    """Parameters for a genre transformation.

    :ivar name: Genre name.
    :ivar conventions: List of genre constraint strings.
    """

    name: str
    conventions: list[str] = Field(default_factory=list)

MoodParams

Bases: BaseModel

Parameters for a mood transformation.

Attributes:

Name	Type	Description
label	str	Free-text mood label.
valence	float	Sentiment polarity in [-1.0, 1.0].
arousal	float	Activation in [0.0, 1.0].

Source code in src/tng/services/transform_service.py

class MoodParams(BaseModel):
    """Parameters for a mood transformation.

    :ivar label: Free-text mood label.
    :ivar valence: Sentiment polarity in [-1.0, 1.0].
    :ivar arousal: Activation in [0.0, 1.0].
    """

    label: str
    valence: float = Field(default=0.0, ge=-1.0, le=1.0)
    arousal: float = Field(default=0.5, ge=0.0, le=1.0)

PovParams

Bases: BaseModel

Parameters for a POV transformation.

Attributes:

Name	Type	Description
focalizer	str	ID of the Character who becomes the focalizer.
distance	FocalizationDistance	Genettean focalization distance.
reliability	ReliabilityLevel	Narrator/focalizer credibility.

Source code in src/tng/services/transform_service.py

class PovParams(BaseModel):
    """Parameters for a POV transformation.

    :ivar focalizer: ID of the Character who becomes the focalizer.
    :ivar distance: Genettean focalization distance.
    :ivar reliability: Narrator/focalizer credibility.
    """

    focalizer: str
    distance: FocalizationDistance = FocalizationDistance.ZERO
    reliability: ReliabilityLevel = ReliabilityLevel.RELIABLE

ReliabilityParams

Bases: BaseModel

Parameters for a reliability transformation.

Attributes:

Name	Type	Description
reliability	ReliabilityLevel	New reliability level for the existing Perspective.

Source code in src/tng/services/transform_service.py

class ReliabilityParams(BaseModel):
    """Parameters for a reliability transformation.

    :ivar reliability: New reliability level for the existing Perspective.
    """

    reliability: ReliabilityLevel

TransformRequest dataclass

Input contract for a transform request.

Parameters:

Name	Type	Description	Default
scene_id	str	Target scene ID.	required
axis	TransformAxis	The transformation axis.	required
parameters	dict[str, Any]	Axis-specific parameter dict.	required
operator	str	Identifier of the requesting user/system.	'system'

Source code in src/tng/services/transform_service.py

@dataclass
class TransformRequest:
    """Input contract for a transform request.

    :param scene_id: Target scene ID.
    :param axis: The transformation axis.
    :param parameters: Axis-specific parameter dict.
    :param operator: Identifier of the requesting user/system.
    """

    scene_id: str
    axis: TransformAxis
    parameters: dict[str, Any]
    operator: str = "system"

TransformResponse dataclass

Result returned to the API after a transform operation.

Parameters:

Name	Type	Description	Default
transform_id	str	ID of the created Transform audit node.	required
scene_id	str	Target scene ID.	required
axis	str	The axis that was applied.	required
produced_id	str	ID of the new state node created.	required
status	str	Always "accepted" on success.	'accepted'

Source code in src/tng/services/transform_service.py

@dataclass
class TransformResponse:
    """Result returned to the API after a transform operation.

    :param transform_id: ID of the created Transform audit node.
    :param scene_id: Target scene ID.
    :param axis: The axis that was applied.
    :param produced_id: ID of the new state node created.
    :param status: Always ``"accepted"`` on success.
    """

    transform_id: str
    scene_id: str
    axis: str
    produced_id: str
    status: str = "accepted"

TransformService

Validates and applies transformation axes.

Parameters:

Name	Type	Description	Default
repo	GraphRepository	Open GraphRepository instance.	required

Source code in src/tng/services/transform_service.py

class TransformService:
    """Validates and applies transformation axes.

    :param repo: Open ``GraphRepository`` instance.
    """

    def __init__(self, repo: GraphRepository) -> None:
        self._repo = repo

    def apply(self, request: TransformRequest) -> TransformResponse:
        """Validate parameters and apply a transformation to a scene.

        :param request: The transform request.
        :returns: A ``TransformResponse`` with the new transform's ID.
        :raises ValueError: When axis parameters fail validation.
        """
        validated_params = self._validate_params(request.axis, request.parameters)
        transform = Transform(
            id=make_id(),
            axis=request.axis,
            operator=request.operator,
            applied_at=datetime.utcnow(),
            parameters=validated_params,
            scene_id=request.scene_id,
        )
        result = self._repo.apply_transform(transform)
        self._repo.update_narrative_status_for_scene(
            request.scene_id, NarrativeStatus.TRANSFORMED
        )
        logger.info(
            "Applied %s transform to scene %s → produced %s",
            request.axis.value,
            request.scene_id,
            result.produced_id,
        )
        return TransformResponse(
            transform_id=result.id,
            scene_id=request.scene_id,
            axis=request.axis.value,
            produced_id=result.produced_id,
        )

    def apply_bulk(
        self,
        narrative_id: str,
        axis: TransformAxis,
        parameters: dict[str, Any],
        operator: str = "system",
    ) -> list[TransformResponse]:
        """Apply a transformation axis to every scene in a narrative.

        Validates parameters against the axis schema before issuing any write.
        Scenes are processed in sequence order.

        :param narrative_id: Target narrative ID.
        :param axis: The transformation axis.
        :param parameters: Axis-specific parameter dict (validated once).
        :param operator: Identifier of the requesting user/system.
        :returns: List of ``TransformResponse`` — one per scene.
        :raises ValueError: When parameters fail axis validation or the
            narrative has no scenes.
        """
        self._validate_params(axis, parameters)
        scene_ids = self._repo.get_scene_ids(narrative_id)
        if not scene_ids:
            raise ValueError(f"No scenes found for narrative {narrative_id!r}.")
        results = []
        for scene_id in scene_ids:
            result = self.apply(
                TransformRequest(
                    scene_id=scene_id,
                    axis=axis,
                    parameters=parameters,
                    operator=operator,
                )
            )
            results.append(result)
        return results

    def get_history(self, scene_id: str) -> list[dict]:
        """Return the transformation history for a scene.

        :param scene_id: The target scene ID.
        :returns: Ordered list of transform audit dicts.
        """
        return self._repo.get_transform_history(scene_id)

    def _validate_params(
        self, axis: TransformAxis, params: dict[str, Any]
    ) -> dict[str, Any]:
        """Validate axis parameters using the axis-specific Pydantic schema.

        :param axis: The transformation axis.
        :param params: Raw parameter dict from the request.
        :returns: Validated parameter dict.
        :raises ValueError: On validation failure.
        """
        schema = _PARAM_SCHEMAS.get(axis)
        if schema is None:
            raise ValueError(f"No parameter schema defined for axis: {axis}")
        model = schema.model_validate(params)
        return model.model_dump()

apply(request)

Validate parameters and apply a transformation to a scene.

Parameters:

Name	Type	Description	Default
request	TransformRequest	The transform request.	required

Returns:

Type	Description
TransformResponse	A TransformResponse with the new transform's ID.

Raises:

Type	Description
ValueError	When axis parameters fail validation.

Source code in src/tng/services/transform_service.py

def apply(self, request: TransformRequest) -> TransformResponse:
    """Validate parameters and apply a transformation to a scene.

    :param request: The transform request.
    :returns: A ``TransformResponse`` with the new transform's ID.
    :raises ValueError: When axis parameters fail validation.
    """
    validated_params = self._validate_params(request.axis, request.parameters)
    transform = Transform(
        id=make_id(),
        axis=request.axis,
        operator=request.operator,
        applied_at=datetime.utcnow(),
        parameters=validated_params,
        scene_id=request.scene_id,
    )
    result = self._repo.apply_transform(transform)
    self._repo.update_narrative_status_for_scene(
        request.scene_id, NarrativeStatus.TRANSFORMED
    )
    logger.info(
        "Applied %s transform to scene %s → produced %s",
        request.axis.value,
        request.scene_id,
        result.produced_id,
    )
    return TransformResponse(
        transform_id=result.id,
        scene_id=request.scene_id,
        axis=request.axis.value,
        produced_id=result.produced_id,
    )

apply_bulk(narrative_id, axis, parameters, operator='system')

Apply a transformation axis to every scene in a narrative.

Validates parameters against the axis schema before issuing any write. Scenes are processed in sequence order.

Parameters:

Name	Type	Description	Default
narrative_id	str	Target narrative ID.	required
axis	TransformAxis	The transformation axis.	required
parameters	dict[str, Any]	Axis-specific parameter dict (validated once).	required
operator	str	Identifier of the requesting user/system.	'system'

Returns:

Type	Description
list[TransformResponse]	List of TransformResponse — one per scene.

Raises:

Type	Description
ValueError	When parameters fail axis validation or the narrative has no scenes.

Source code in src/tng/services/transform_service.py

def apply_bulk(
    self,
    narrative_id: str,
    axis: TransformAxis,
    parameters: dict[str, Any],
    operator: str = "system",
) -> list[TransformResponse]:
    """Apply a transformation axis to every scene in a narrative.

    Validates parameters against the axis schema before issuing any write.
    Scenes are processed in sequence order.

    :param narrative_id: Target narrative ID.
    :param axis: The transformation axis.
    :param parameters: Axis-specific parameter dict (validated once).
    :param operator: Identifier of the requesting user/system.
    :returns: List of ``TransformResponse`` — one per scene.
    :raises ValueError: When parameters fail axis validation or the
        narrative has no scenes.
    """
    self._validate_params(axis, parameters)
    scene_ids = self._repo.get_scene_ids(narrative_id)
    if not scene_ids:
        raise ValueError(f"No scenes found for narrative {narrative_id!r}.")
    results = []
    for scene_id in scene_ids:
        result = self.apply(
            TransformRequest(
                scene_id=scene_id,
                axis=axis,
                parameters=parameters,
                operator=operator,
            )
        )
        results.append(result)
    return results

get_history(scene_id)

Return the transformation history for a scene.

Parameters:

Name	Type	Description	Default
scene_id	str	The target scene ID.	required

Returns:

Type	Description
list[dict]	Ordered list of transform audit dicts.

Source code in src/tng/services/transform_service.py

def get_history(self, scene_id: str) -> list[dict]:
    """Return the transformation history for a scene.

    :param scene_id: The target scene ID.
    :returns: Ordered list of transform audit dicts.
    """
    return self._repo.get_transform_history(scene_id)

tng.services.pattern_service

Pattern service — detects and instantiates narrative pattern templates.

Implements SRS §6.2 (Diagram 8): pattern detection is a matching operation between incoming atoms/events and a library of Pattern templates stored in the graph.

Architecture

• The service holds a reference to the GraphRepository for template lookups and instance persistence.
• Pattern matching is a Strategy: each PatternMatcher checks a single template against the atom/event set and returns an optional PatternInstance with a confidence score.
• Multiple matchers can fire on the same scene; overlapping patterns are not collapsed — they are represented as separate PatternInstance nodes (SRS §6.2).
• Pattern templates are loaded from the graph at startup and cached in memory for the lifetime of the service.

Built-in matchers

The service ships four lightweight matchers that cover common patterns from the SRS seed data. Additional matchers can be registered at startup without modifying this module.

KeywordAtomMatcher dataclass

Matches a pattern by searching atom texts for keyword strings.

Parameters:

Name	Type	Description	Default
keywords	frozenset[str]	Set of lowercase keywords to search for in atom text.	required
base_confidence	float	Confidence assigned on single keyword match.	0.7

Source code in src/tng/services/pattern_service.py

@dataclass
class KeywordAtomMatcher:
    """Matches a pattern by searching atom texts for keyword strings.

    :param keywords: Set of lowercase keywords to search for in atom text.
    :param base_confidence: Confidence assigned on single keyword match.
    """

    keywords: frozenset[str]
    base_confidence: float = 0.70

    def match(
        self,
        template: Pattern,
        atoms: list[Atom],
        events: list[Event],
    ) -> PatternInstance | None:
        """Return a ``PatternInstance`` if keywords are found in atom text.

        :param template: The template to match.
        :param atoms: Scene atoms whose text is scanned.
        :param events: Scene events (not used by this matcher).
        :returns: A ``PatternInstance`` or ``None``.
        """
        matched_atoms = [
            a
            for a in atoms
            if any(kw in a.text.lower() for kw in self.keywords)
        ]
        if not matched_atoms:
            return None
        confidence = min(0.95, self.base_confidence + 0.04 * len(matched_atoms))
        return PatternInstance(
            id=make_id(),
            slot="scene-core",
            confidence=confidence,
            template=template,
            realized_atom_ids=[a.id for a in matched_atoms],
        )

match(template, atoms, events)

Return a PatternInstance if keywords are found in atom text.

Parameters:

Name	Type	Description	Default
template	Pattern	The template to match.	required
atoms	list[Atom]	Scene atoms whose text is scanned.	required
events	list[Event]	Scene events (not used by this matcher).	required

Returns:

Type	Description
PatternInstance | None	A PatternInstance or None.

Source code in src/tng/services/pattern_service.py

def match(
    self,
    template: Pattern,
    atoms: list[Atom],
    events: list[Event],
) -> PatternInstance | None:
    """Return a ``PatternInstance`` if keywords are found in atom text.

    :param template: The template to match.
    :param atoms: Scene atoms whose text is scanned.
    :param events: Scene events (not used by this matcher).
    :returns: A ``PatternInstance`` or ``None``.
    """
    matched_atoms = [
        a
        for a in atoms
        if any(kw in a.text.lower() for kw in self.keywords)
    ]
    if not matched_atoms:
        return None
    confidence = min(0.95, self.base_confidence + 0.04 * len(matched_atoms))
    return PatternInstance(
        id=make_id(),
        slot="scene-core",
        confidence=confidence,
        template=template,
        realized_atom_ids=[a.id for a in matched_atoms],
    )

PatternMatcher

Bases: Protocol

Contract for a single-pattern matching strategy.

:method match: Test atoms and events against a template. Return a PatternInstance if matched, else None.

Source code in src/tng/services/pattern_service.py

class PatternMatcher(Protocol):
    """Contract for a single-pattern matching strategy.

    :method match: Test ``atoms`` and ``events`` against a template.
        Return a ``PatternInstance`` if matched, else ``None``.
    """

    def match(
        self,
        template: Pattern,
        atoms: list[Atom],
        events: list[Event],
    ) -> PatternInstance | None:
        """Attempt to match the template against the provided atoms/events.

        :param template: The pattern template to match against.
        :param atoms: Scene atoms available for slot binding.
        :param events: Scene events available for slot binding.
        :returns: A ``PatternInstance`` on match, or ``None``.
        """
        ...

match(template, atoms, events)

Attempt to match the template against the provided atoms/events.

Parameters:

Name	Type	Description	Default
template	Pattern	The pattern template to match against.	required
atoms	list[Atom]	Scene atoms available for slot binding.	required
events	list[Event]	Scene events available for slot binding.	required

Returns:

Type	Description
PatternInstance | None	A PatternInstance on match, or None.

Source code in src/tng/services/pattern_service.py

def match(
    self,
    template: Pattern,
    atoms: list[Atom],
    events: list[Event],
) -> PatternInstance | None:
    """Attempt to match the template against the provided atoms/events.

    :param template: The pattern template to match against.
    :param atoms: Scene atoms available for slot binding.
    :param events: Scene events available for slot binding.
    :returns: A ``PatternInstance`` on match, or ``None``.
    """
    ...

PatternService

Detects narrative patterns in a set of atoms and events.

Parameters:

Name	Type	Description	Default
repo	GraphRepository	GraphRepository for template lookups.	required
matchers	dict[str, PatternMatcher] | None	Matcher registry mapping pattern ID → matcher. Defaults to DEFAULT_MATCHERS.	None

Source code in src/tng/services/pattern_service.py

class PatternService:
    """Detects narrative patterns in a set of atoms and events.

    :param repo: ``GraphRepository`` for template lookups.
    :param matchers: Matcher registry mapping pattern ID → matcher.
        Defaults to ``DEFAULT_MATCHERS``.
    """

    def __init__(
        self,
        repo: GraphRepository,
        matchers: dict[str, PatternMatcher] | None = None,
    ) -> None:
        self._repo = repo
        self._matchers = matchers if matchers is not None else DEFAULT_MATCHERS.copy()

    def detect_patterns(
        self,
        atoms: list[Atom],
        events: list[Event],
        narrative_id: str,
    ) -> list[PatternInstance]:
        """Run all registered matchers against the given atoms and events.

        Pattern templates are fetched from the graph by ID.  If a template
        does not exist in the graph it is silently skipped (not auto-created).

        :param atoms: Scene atoms to match against.
        :param events: Scene events to match against.
        :param narrative_id: ID of the parent narrative (for logging).
        :returns: List of ``PatternInstance`` objects (may be empty).
        """
        instances: list[PatternInstance] = []
        for pattern_id, matcher in self._matchers.items():
            template = self._repo.get_pattern(pattern_id)
            if template is None:
                logger.debug("Pattern template %r not in graph; skipping.", pattern_id)
                continue
            instance = matcher.match(template, atoms, events)
            if instance:
                instances.append(instance)
        return instances

    def register_pattern(self, pattern: Pattern, matcher: PatternMatcher) -> None:
        """Register a new pattern template and its matcher at runtime.

        The pattern template is persisted to the graph immediately.

        :param pattern: The ``Pattern`` template to add to the library.
        :param matcher: The matcher strategy for this pattern.
        """
        self._repo.save_pattern(pattern)
        self._matchers[pattern.id] = matcher
        logger.info("Registered pattern %r (%s).", pattern.id, pattern.name)

detect_patterns(atoms, events, narrative_id)

Run all registered matchers against the given atoms and events.

Pattern templates are fetched from the graph by ID. If a template does not exist in the graph it is silently skipped (not auto-created).

Parameters:

Name	Type	Description	Default
atoms	list[Atom]	Scene atoms to match against.	required
events	list[Event]	Scene events to match against.	required
narrative_id	str	ID of the parent narrative (for logging).	required

Returns:

Type	Description
list[PatternInstance]	List of PatternInstance objects (may be empty).

Source code in src/tng/services/pattern_service.py

def detect_patterns(
    self,
    atoms: list[Atom],
    events: list[Event],
    narrative_id: str,
) -> list[PatternInstance]:
    """Run all registered matchers against the given atoms and events.

    Pattern templates are fetched from the graph by ID.  If a template
    does not exist in the graph it is silently skipped (not auto-created).

    :param atoms: Scene atoms to match against.
    :param events: Scene events to match against.
    :param narrative_id: ID of the parent narrative (for logging).
    :returns: List of ``PatternInstance`` objects (may be empty).
    """
    instances: list[PatternInstance] = []
    for pattern_id, matcher in self._matchers.items():
        template = self._repo.get_pattern(pattern_id)
        if template is None:
            logger.debug("Pattern template %r not in graph; skipping.", pattern_id)
            continue
        instance = matcher.match(template, atoms, events)
        if instance:
            instances.append(instance)
    return instances

register_pattern(pattern, matcher)

Register a new pattern template and its matcher at runtime.

The pattern template is persisted to the graph immediately.

Parameters:

Name	Type	Description	Default
pattern	Pattern	The Pattern template to add to the library.	required
matcher	PatternMatcher	The matcher strategy for this pattern.	required

Source code in src/tng/services/pattern_service.py

def register_pattern(self, pattern: Pattern, matcher: PatternMatcher) -> None:
    """Register a new pattern template and its matcher at runtime.

    The pattern template is persisted to the graph immediately.

    :param pattern: The ``Pattern`` template to add to the library.
    :param matcher: The matcher strategy for this pattern.
    """
    self._repo.save_pattern(pattern)
    self._matchers[pattern.id] = matcher
    logger.info("Registered pattern %r (%s).", pattern.id, pattern.name)

VerbFamilyMatcher dataclass

Matches a pattern whose template name is associated with a set of verbs.

Parameters:

Name	Type	Description	Default
verb_hints	frozenset[str]	Set of lowercase verb strings that signal this pattern.	required
base_confidence	float	Confidence assigned when a hint verb is found.	0.75

Source code in src/tng/services/pattern_service.py

@dataclass
class VerbFamilyMatcher:
    """Matches a pattern whose template name is associated with a set of verbs.

    :param verb_hints: Set of lowercase verb strings that signal this pattern.
    :param base_confidence: Confidence assigned when a hint verb is found.
    """

    verb_hints: frozenset[str]
    base_confidence: float = 0.75

    def match(
        self,
        template: Pattern,
        atoms: list[Atom],
        events: list[Event],
    ) -> PatternInstance | None:
        """Return a ``PatternInstance`` if any event verb matches a hint.

        :param template: The template to match.
        :param atoms: Scene atoms (not used by this matcher).
        :param events: Scene events whose verbs are checked.
        :returns: A ``PatternInstance`` or ``None``.
        """
        matched_events = [e for e in events if e.verb in self.verb_hints]
        if not matched_events:
            return None
        confidence = min(
            0.95, self.base_confidence + 0.05 * (len(matched_events) - 1)
        )
        instance = PatternInstance(
            id=make_id(),
            slot="scene-core",
            confidence=confidence,
            template=template,
            realized_event_ids=[e.id for e in matched_events],
        )
        logger.debug(
            "Pattern %r matched with confidence %.2f", template.name, confidence
        )
        return instance

match(template, atoms, events)

Return a PatternInstance if any event verb matches a hint.

Parameters:

Name	Type	Description	Default
template	Pattern	The template to match.	required
atoms	list[Atom]	Scene atoms (not used by this matcher).	required
events	list[Event]	Scene events whose verbs are checked.	required

Returns:

Type	Description
PatternInstance | None	A PatternInstance or None.

Source code in src/tng/services/pattern_service.py

def match(
    self,
    template: Pattern,
    atoms: list[Atom],
    events: list[Event],
) -> PatternInstance | None:
    """Return a ``PatternInstance`` if any event verb matches a hint.

    :param template: The template to match.
    :param atoms: Scene atoms (not used by this matcher).
    :param events: Scene events whose verbs are checked.
    :returns: A ``PatternInstance`` or ``None``.
    """
    matched_events = [e for e in events if e.verb in self.verb_hints]
    if not matched_events:
        return None
    confidence = min(
        0.95, self.base_confidence + 0.05 * (len(matched_events) - 1)
    )
    instance = PatternInstance(
        id=make_id(),
        slot="scene-core",
        confidence=confidence,
        template=template,
        realized_event_ids=[e.id for e in matched_events],
    )
    logger.debug(
        "Pattern %r matched with confidence %.2f", template.name, confidence
    )
    return instance

tng.services.render_service

Render service — dispatches render requests to registered renderers.

The service maintains a registry of RendererProtocol implementations keyed by RenderType. Renderers are registered at startup; new output types can be added without modifying this module (open/closed principle).

No renderer is allowed to issue Cypher directly. All graph data arrives as a GraphState snapshot fetched by this service via the repository.

RenderService

Dispatches render requests to the appropriate renderer implementation.

Parameters:

Name	Type	Description	Default
repo	GraphRepository	Open GraphRepository for fetching graph state.	required
renderers	dict[RenderType, RendererProtocol] | None	Renderer registry. Defaults to the five built-in renderers; inject custom implementations for extensibility.	None

Source code in src/tng/services/render_service.py

class RenderService:
    """Dispatches render requests to the appropriate renderer implementation.

    :param repo: Open ``GraphRepository`` for fetching graph state.
    :param renderers: Renderer registry.  Defaults to the five built-in
        renderers; inject custom implementations for extensibility.
    """

    def __init__(
        self,
        repo: GraphRepository,
        renderers: dict[RenderType, RendererProtocol] | None = None,
    ) -> None:
        self._repo = repo
        self._renderers = renderers or _DEFAULT_RENDERERS.copy()

    def render(
        self,
        narrative_id: str,
        render_type: RenderType,
        params: dict[str, Any] | None = None,
    ) -> RenderOutput:
        """Fetch graph state and render it to the requested format.

        :param narrative_id: The narrative to render.
        :param render_type: The desired output format.
        :param params: Renderer-specific parameters (optional).
        :returns: A ``RenderOutput`` containing the rendered content.
        :raises KeyError: When ``render_type`` has no registered renderer.
        :raises ValueError: When the narrative does not exist.
        """
        renderer = self._renderers.get(render_type)
        if renderer is None:
            raise KeyError(f"No renderer registered for type: {render_type}")

        graph_state = self._repo.get_graph_state(narrative_id)
        if graph_state is None:
            raise ValueError(f"Narrative not found: {narrative_id}")

        self._repo.update_narrative_status(narrative_id, NarrativeStatus.RENDERED)
        logger.info(
            "Rendering narrative %s as %s.", narrative_id, render_type.value
        )
        return renderer.render(graph_state, params or {})

    def register_renderer(
        self, render_type: RenderType, renderer: RendererProtocol
    ) -> None:
        """Register or replace a renderer at runtime.

        :param render_type: The output format key.
        :param renderer: The renderer implementation to register.
        """
        self._renderers[render_type] = renderer
        logger.info("Registered renderer for %s.", render_type.value)

register_renderer(render_type, renderer)

Register or replace a renderer at runtime.

Parameters:

Name	Type	Description	Default
render_type	RenderType	The output format key.	required
renderer	RendererProtocol	The renderer implementation to register.	required

Source code in src/tng/services/render_service.py

def register_renderer(
    self, render_type: RenderType, renderer: RendererProtocol
) -> None:
    """Register or replace a renderer at runtime.

    :param render_type: The output format key.
    :param renderer: The renderer implementation to register.
    """
    self._renderers[render_type] = renderer
    logger.info("Registered renderer for %s.", render_type.value)

render(narrative_id, render_type, params=None)

Fetch graph state and render it to the requested format.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative to render.	required
render_type	RenderType	The desired output format.	required
params	dict[str, Any] | None	Renderer-specific parameters (optional).	None

Returns:

Type	Description
RenderOutput	A RenderOutput containing the rendered content.

Raises:

Type	Description
KeyError	When render_type has no registered renderer.
ValueError	When the narrative does not exist.

Source code in src/tng/services/render_service.py

def render(
    self,
    narrative_id: str,
    render_type: RenderType,
    params: dict[str, Any] | None = None,
) -> RenderOutput:
    """Fetch graph state and render it to the requested format.

    :param narrative_id: The narrative to render.
    :param render_type: The desired output format.
    :param params: Renderer-specific parameters (optional).
    :returns: A ``RenderOutput`` containing the rendered content.
    :raises KeyError: When ``render_type`` has no registered renderer.
    :raises ValueError: When the narrative does not exist.
    """
    renderer = self._renderers.get(render_type)
    if renderer is None:
        raise KeyError(f"No renderer registered for type: {render_type}")

    graph_state = self._repo.get_graph_state(narrative_id)
    if graph_state is None:
        raise ValueError(f"Narrative not found: {narrative_id}")

    self._repo.update_narrative_status(narrative_id, NarrativeStatus.RENDERED)
    logger.info(
        "Rendering narrative %s as %s.", narrative_id, render_type.value
    )
    return renderer.render(graph_state, params or {})

tng.repository.graph_repository

Graph repository — single point of contact with Neo4j.

All Cypher queries are executed here and nowhere else. Service-layer code receives domain objects; it never sees raw driver records. This enforces the boundary described in SRS §3.2 (Diagram 2).

Connection lifecycle

The repository accepts a neo4j.Driver instance. The driver manages its own connection pool; one driver per process is the correct pattern. The close() method must be called at application shutdown to release pool resources.

Transaction discipline

• Writes — all multi-statement writes use managed transactions (session.execute_write) so that failures roll back atomically.
• Small reads — bounded single-record reads use execute_query.
• Streaming reads — render queries that may return many rows use session.run with lazy cursor iteration.

GraphRepository

Abstracts all Neo4j interactions for the TNGS application.

Parameters:

Name	Type	Description	Default
driver	Driver	An authenticated neo4j.Driver instance.	required
database	str	Name of the Neo4j database to target.	'neo4j'

Source code in src/tng/repository/graph_repository.py

class GraphRepository:
    """Abstracts all Neo4j interactions for the TNGS application.

    :param driver: An authenticated ``neo4j.Driver`` instance.
    :param database: Name of the Neo4j database to target.
    """

    def __init__(self, driver: Driver, database: str = "neo4j") -> None:
        self._driver = driver
        self._db = database

    # ── Lifecycle ─────────────────────────────────────────────────────────────

    def close(self) -> None:
        """Release driver connection pool resources."""
        self._driver.close()

    # ── Health ────────────────────────────────────────────────────────────────

    def ping(self) -> bool:
        """Return True when Neo4j is reachable and the database responds.

        :returns: ``True`` on success.
        :raises Exception: On Neo4j connectivity failure — callers decide how to
            handle it.
        """
        result = self._driver.execute_query(
            Q.HEALTH_PING, database_=self._db
        )
        return bool(result.records)

    # ── Schema bootstrap ──────────────────────────────────────────────────────

    def apply_schema(self) -> None:
        """Apply all constraints and indexes idempotently.

        Safe to call on every startup; uses ``IF NOT EXISTS`` guards.
        """
        with self._driver.session(database=self._db) as session:
            for stmt in Q.SCHEMA_CONSTRAINTS + Q.SCHEMA_INDEXES:
                session.run(stmt)
        logger.info("Schema constraints and indexes verified.")

    # ── Narrative ─────────────────────────────────────────────────────────────

    def save_narrative(self, narrative: Narrative) -> Narrative:
        """Persist or update a Narrative node.

        :param narrative: The domain Narrative to persist.
        :returns: The same narrative (unchanged — the graph write is
            idempotent via MERGE).
        """

        def _write(tx: Any) -> None:
            tx.run(
                Q.MERGE_NARRATIVE,
                id=narrative.id,
                title=narrative.title,
                status=narrative.status.value,
                source_ref=narrative.source_ref,
                created_at=_dt_str(narrative.created_at),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        logger.debug("Saved narrative %s", narrative.id)
        return narrative

    def get_narrative(self, narrative_id: str) -> Narrative | None:
        """Retrieve a Narrative by ID with all nested scenes and atoms.

        :param narrative_id: The narrative's unique identifier.
        :returns: A populated ``Narrative`` or ``None`` if not found.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_NARRATIVE, id=narrative_id, database_=self._db
        )
        if not records:
            return None
        node = records[0]["n"]
        narrative = Narrative(
            id=node["id"],
            title=node["title"],
            status=NarrativeStatus(node.get("status", "draft")),
            source_ref=node.get("source_ref", ""),
            created_at=_parse_dt(node.get("created_at")),
        )
        narrative.scenes = self._get_scenes(narrative_id)
        return narrative

    def update_narrative_status(
        self, narrative_id: str, status: NarrativeStatus
    ) -> None:
        """Update the status property on a Narrative node.

        :param narrative_id: Target narrative ID.
        :param status: New status value.
        """

        def _write(tx: Any) -> None:
            tx.run(
                Q.UPDATE_NARRATIVE_STATUS,
                id=narrative_id,
                status=status.value,
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)

    def update_narrative_status_for_scene(
        self, scene_id: str, status: NarrativeStatus
    ) -> None:
        """Update the status of the Narrative that contains a given scene.

        :param scene_id: ID of a scene whose parent narrative should be updated.
        :param status: New status value.
        """

        def _write(tx: Any) -> None:
            tx.run(
                """
                MATCH (n:Narrative)-[:HAS_SCENE]->(s:Scene {id: $scene_id})
                SET n.status = $status
                """,
                scene_id=scene_id,
                status=status.value,
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)

    def archive_narrative(self, narrative_id: str) -> bool:
        """Set a Narrative's status to archived.

        :param narrative_id: Target narrative ID.
        :returns: ``True`` if the narrative was found and archived.
        """

        def _write(tx: Any) -> Any:
            return tx.run(Q.ARCHIVE_NARRATIVE, id=narrative_id).single()

        with self._driver.session(database=self._db) as session:
            result = session.execute_write(_write)
        return result is not None

    # ── Scene ─────────────────────────────────────────────────────────────────

    def save_scene(self, scene: Scene, narrative_id: str) -> Scene:
        """Persist a Scene and link it to its parent Narrative.

        :param scene: The domain Scene to persist.
        :param narrative_id: The parent narrative's ID.
        :returns: The saved scene.
        """

        def _write(tx: Any) -> None:
            tx.run(
                Q.MERGE_SCENE,
                id=scene.id,
                sequence=scene.sequence,
                summary=scene.summary,
                narrative_id=narrative_id,
            )
            for atom in scene.atoms:
                self._save_atom_tx(tx, atom, scene.id)
            for event in scene.events:
                self._save_event_tx(tx, event, scene.id)
            for instance in scene.pattern_instances:
                self._save_pattern_instance_tx(tx, instance, scene.id)

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        return scene

    def _get_scenes(self, narrative_id: str) -> list[Scene]:
        records, _, _ = self._driver.execute_query(
            Q.GET_SCENES_FOR_NARRATIVE,
            narrative_id=narrative_id,
            database_=self._db,
        )
        scenes = []
        for rec in records:
            node = rec["s"]
            scene = Scene(
                id=node["id"],
                sequence=node.get("sequence", 1),
                summary=node.get("summary", ""),
            )
            scene.atoms = self._get_atoms(scene.id)
            scenes.append(scene)
        return scenes

    # ── Atom ──────────────────────────────────────────────────────────────────

    def get_scene_ids(self, narrative_id: str) -> list[str]:
        """Return scene IDs for a narrative in sequence order.

        :param narrative_id: The narrative's unique identifier.
        :returns: List of scene ID strings, ordered by sequence.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_SCENE_IDS_FOR_NARRATIVE,
            narrative_id=narrative_id,
            database_=self._db,
        )
        return [rec["scene_id"] for rec in records]

    def _save_atom_tx(self, tx: Any, atom: Atom, scene_id: str) -> None:
        tx.run(
            Q.MERGE_ATOM,
            id=atom.id,
            text=atom.text,
            kind=atom.kind.value,
            surface_order=atom.surface_order,
            confidence=atom.confidence,
            needs_review=atom.needs_review,
            scene_id=scene_id,
        )
        for tag in atom.code_tags:
            tx.run(
                Q.MERGE_CODE_TAG,
                id=tag.id,
                code=tag.code.value,
                label=tag.label,
                atom_id=atom.id,
            )

    def _get_atoms(self, scene_id: str) -> list[Atom]:
        records, _, _ = self._driver.execute_query(
            Q.GET_ATOMS_FOR_SCENE, scene_id=scene_id, database_=self._db
        )
        atoms = []
        for rec in records:
            node = rec["a"]
            from tng.domain.enums import AtomKind
            atoms.append(
                Atom(
                    id=node["id"],
                    text=rec["resolved_text"],
                    kind=AtomKind(node.get("kind", "descriptive")),
                    surface_order=node.get("surface_order", 0),
                    confidence=node.get("confidence", 1.0),
                    needs_review=node.get("needs_review", False),
                )
            )
        return atoms

    # ── Event ─────────────────────────────────────────────────────────────────

    def _save_event_tx(self, tx: Any, event: Event, scene_id: str) -> None:
        tx.run(
            Q.MERGE_EVENT,
            id=event.id,
            verb=event.verb,
            tense=event.tense,
            aspect=event.aspect,
            confidence=event.confidence,
            needs_review=event.needs_review,
            scene_id=scene_id,
        )
        for char in event.participants:
            tx.run(
                Q.MERGE_CHARACTER,
                id=char.id,
                name=char.name,
                role=char.role,
            )
            tx.run(
                Q.LINK_CHARACTER_TO_EVENT,
                character_id=char.id,
                event_id=event.id,
            )

    # ── Pattern ───────────────────────────────────────────────────────────────

    def save_pattern(self, pattern: Pattern) -> Pattern:
        """Persist a Pattern template.

        :param pattern: The pattern to save.
        :returns: The same pattern (unchanged).
        """

        def _write(tx: Any) -> None:
            tx.run(
                Q.MERGE_PATTERN,
                id=pattern.id,
                name=pattern.name,
                family=pattern.family,
                description=pattern.description,
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        return pattern

    def get_pattern(self, pattern_id: str) -> Pattern | None:
        """Retrieve a single Pattern template by ID.

        :param pattern_id: The pattern's unique identifier.
        :returns: A ``Pattern`` or ``None`` if not found.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_PATTERN, id=pattern_id, database_=self._db
        )
        if not records:
            return None
        node = records[0]["p"]
        return Pattern(
            id=node["id"],
            name=node["name"],
            family=node["family"],
            description=node.get("description", ""),
        )

    def list_patterns(self, family: str | None = None) -> list[Pattern]:
        """List pattern templates, optionally filtered by family.

        :param family: If provided, only return patterns of this family.
        :returns: List of matching patterns.
        """
        records, _, _ = self._driver.execute_query(
            Q.LIST_PATTERNS, family=family, database_=self._db
        )
        return [
            Pattern(
                id=r["p"]["id"],
                name=r["p"]["name"],
                family=r["p"]["family"],
                description=r["p"].get("description", ""),
            )
            for r in records
        ]

    def _save_pattern_instance_tx(
        self, tx: Any, instance: PatternInstance, scene_id: str
    ) -> None:
        if instance.template is None:
            return
        tx.run(
            Q.MERGE_PATTERN_INSTANCE,
            id=instance.id,
            slot=instance.slot,
            confidence=instance.confidence,
            needs_review=instance.needs_review,
            scene_id=scene_id,
            pattern_id=instance.template.id,
        )
        for atom_id in instance.realized_atom_ids:
            tx.run(
                Q.LINK_INSTANCE_TO_ATOM,
                instance_id=instance.id,
                atom_id=atom_id,
            )
        for event_id in instance.realized_event_ids:
            tx.run(
                Q.LINK_INSTANCE_TO_EVENT,
                instance_id=instance.id,
                event_id=event_id,
            )

    def list_pattern_instances(self, narrative_id: str) -> list[dict[str, Any]]:
        """List all PatternInstances for a narrative with context.

        :param narrative_id: The narrative to query.
        :returns: List of dicts with instance, pattern, and scene_id.
        """
        records, _, _ = self._driver.execute_query(
            Q.LIST_PATTERN_INSTANCES,
            narrative_id=narrative_id,
            database_=self._db,
        )
        return [
            {
                "instance_id": r["pi"]["id"],
                "slot": r["pi"]["slot"],
                "confidence": r["pi"]["confidence"],
                "pattern_id": r["p"]["id"],
                "pattern_name": r["p"]["name"],
                "pattern_family": r["p"]["family"],
                "scene_id": r["scene_id"],
            }
            for r in records
        ]

    # ── Transforms ────────────────────────────────────────────────────────────

    def apply_transform(self, transform: Transform) -> Transform:
        """Dispatch and persist a transformation on a scene.

        Routes to the appropriate axis-specific Cypher query.  The old
        state node is detached (not deleted) and the new state node is
        created in a single managed transaction.

        :param transform: Fully populated Transform domain object.
        :returns: The same transform (with ``produced_id`` set if applicable).
        :raises ValueError: For unknown or unsupported axis values.
        """
        axis = transform.axis
        dispatch = {
            TransformAxis.POV: self._apply_pov,
            TransformAxis.MOOD: self._apply_mood,
            TransformAxis.GENRE: self._apply_genre,
            TransformAxis.CHRONOTOPE: self._apply_chronotope,
            TransformAxis.RELIABILITY: self._apply_reliability,
            TransformAxis.CODE_OVERLAY: self._apply_code_overlay,
        }
        handler = dispatch.get(axis)
        if handler is None:
            raise ValueError(f"Unknown transform axis: {axis}")
        return handler(transform)

    def _apply_pov(self, transform: Transform) -> Transform:
        p = transform.parameters
        perspective_id = f"pov-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_POV_TRANSFORM,
                scene_id=transform.scene_id,
                perspective_id=perspective_id,
                focalizer=p.get("focalizer", ""),
                distance=p.get("distance", "zero"),
                reliability=p.get("reliability", "reliable"),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = perspective_id
        return transform

    def _apply_mood(self, transform: Transform) -> Transform:
        p = transform.parameters
        mood_id = f"mood-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_MOOD_TRANSFORM,
                scene_id=transform.scene_id,
                mood_id=mood_id,
                label=p.get("label", "neutral"),
                valence=float(p.get("valence", 0.0)),
                arousal=float(p.get("arousal", 0.5)),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = mood_id
        return transform

    def _apply_genre(self, transform: Transform) -> Transform:
        p = transform.parameters
        genre_id = f"genre-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_GENRE_TRANSFORM,
                scene_id=transform.scene_id,
                genre_id=genre_id,
                name=p.get("name", ""),
                conventions=json.dumps(p.get("conventions", [])),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = genre_id
        return transform

    def _apply_chronotope(self, transform: Transform) -> Transform:
        p = transform.parameters
        chronotope_id = f"ct-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_CHRONOTOPE_TRANSFORM,
                scene_id=transform.scene_id,
                chronotope_id=chronotope_id,
                time_mode=p.get("time_mode", "linear"),
                space_mode=p.get("space_mode", "bounded"),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = chronotope_id
        return transform

    def _apply_reliability(self, transform: Transform) -> Transform:
        p = transform.parameters
        perspective_id = f"pov-rel-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_RELIABILITY_TRANSFORM,
                scene_id=transform.scene_id,
                perspective_id=perspective_id,
                reliability=p.get("reliability", "reliable"),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = perspective_id
        return transform

    def _apply_code_overlay(self, transform: Transform) -> Transform:
        p = transform.parameters
        tag_id = f"tag-{transform.id}"

        def _write(tx: Any) -> None:
            tx.run(
                Q.APPLY_CODE_OVERLAY_TRANSFORM,
                tag_id=tag_id,
                code=p.get("code", "semic"),
                label=p.get("label", ""),
                atom_id=p.get("atom_id", ""),
                transform_id=transform.id,
                operator=transform.operator,
                applied_at=_dt_str(transform.applied_at),
                parameters=json.dumps(p),
            )

        with self._driver.session(database=self._db) as session:
            session.execute_write(_write)
        transform.produced_id = tag_id
        return transform

    def get_transform(self, transform_id: str) -> dict[str, Any] | None:
        """Retrieve a Transform audit record by ID.

        :param transform_id: The transform's unique identifier.
        :returns: A dict with transform details or ``None`` if not found.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_TRANSFORM, id=transform_id, database_=self._db
        )
        if not records:
            return None
        r = records[0]
        return {
            "id": transform_id,
            "scene_id": r.get("scene_id"),
            "produced_type": r.get("produced_type"),
            "produced_id": r.get("produced_id"),
        }

    def get_transform_history(self, scene_id: str) -> list[dict[str, Any]]:
        """Return the ordered transformation history for a scene.

        :param scene_id: The scene to query.
        :returns: List of transform dicts ordered by applied_at ASC.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_TRANSFORM_HISTORY, scene_id=scene_id, database_=self._db
        )
        return [dict(r) for r in records]

    # ── Render support ────────────────────────────────────────────────────────

    def get_atoms_with_context(self, narrative_id: str) -> list[dict[str, Any]]:
        """Fetch atoms in surface order with their current scene context.

        Used by the prose renderer.

        :param narrative_id: The narrative to render.
        :returns: Row dicts with atom text, scene metadata, and perspective/mood.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_ATOMS_WITH_CONTEXT,
            narrative_id=narrative_id,
            database_=self._db,
        )
        return [dict(r) for r in records]

    def get_graph_state(self, narrative_id: str) -> GraphState | None:
        """Return a complete in-memory snapshot of a narrative's graph state.

        :param narrative_id: The narrative to snapshot.
        :returns: A ``GraphState`` or ``None`` if the narrative doesn't exist.
        """
        narrative = self.get_narrative(narrative_id)
        if narrative is None:
            return None
        transforms_raw = []
        for scene in narrative.scenes:
            transforms_raw.extend(self.get_transform_history(scene.id))

        transforms = [
            Transform(
                id=r["id"],
                axis=TransformAxis(r["axis"]),
                operator=r.get("operator", "system"),
                applied_at=_parse_dt(r.get("applied_at")),
                parameters=json.loads(r.get("parameters") or "{}"),
                scene_id=r.get("scene_id", ""),
                produced_id=r.get("produced_id", ""),
            )
            for r in transforms_raw
        ]
        event_relations = self.get_event_relations(narrative_id)
        return GraphState(
            narrative=narrative,
            transforms=transforms,
            event_relations=event_relations,
        )

    # ── Atom revisions ────────────────────────────────────────────────────────

    def revise_atom(
        self,
        atom_id: str,
        revision_id: str,
        text: str,
        revised_at: datetime,
        operator: str,
        reason: str,
    ) -> bool:
        """Create an AtomRevision node and re-point CURRENT_REVISION.

        The old CURRENT_REVISION edge is detached (not deleted) and a
        SUPERSEDES edge is added from the new revision to the old one.

        :param atom_id: The target Atom's ID.
        :param revision_id: Pre-generated ID for the new AtomRevision node.
        :param text: Revised prose text.
        :param revised_at: UTC timestamp.
        :param operator: Identifier of the requesting user/system.
        :param reason: Optional reason for the change.
        :returns: ``True`` if the atom was found and revised.
        """

        def _write(tx: Any) -> Any:
            return tx.run(
                Q.CREATE_ATOM_REVISION,
                atom_id=atom_id,
                revision_id=revision_id,
                text=text,
                revised_at=_dt_str(revised_at),
                operator=operator,
                reason=reason,
            ).single()

        with self._driver.session(database=self._db) as session:
            result = session.execute_write(_write)
        return result is not None

    def get_atom_revisions(self, atom_id: str) -> list[dict[str, Any]]:
        """Return all AtomRevision nodes for an atom, oldest first.

        :param atom_id: The target Atom's ID.
        :returns: List of revision dicts with ``id``, ``text``, ``revised_at``,
            ``operator``, and ``reason`` keys.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_ATOM_REVISIONS, atom_id=atom_id, database_=self._db
        )
        return [
            {
                "id": rec["r"]["id"],
                "atom_id": rec["r"]["atom_id"],
                "text": rec["r"]["text"],
                "revised_at": rec["r"]["revised_at"],
                "operator": rec["r"].get("operator", "system"),
                "reason": rec["r"].get("reason", ""),
            }
            for rec in records
        ]

    def get_event_relations(self, narrative_id: str) -> list[EventRelation]:
        """Fetch all inter-event causal and temporal relationships.

        Returns ``CAUSES``, ``ENABLES``, ``PREVENTS``, and ``PRECEDES``
        edges between events within the narrative's scenes.  Used by the
        GraphML renderer to draw and tension-score causal graph structure.

        :param narrative_id: The narrative to query.
        :returns: List of ``EventRelation`` objects.
        """
        records, _, _ = self._driver.execute_query(
            Q.GET_EVENT_RELATIONS,
            narrative_id=narrative_id,
            database_=self._db,
        )
        return [
            EventRelation(
                source_id=r["source_id"],
                target_id=r["target_id"],
                relation_type=r["relation_type"],
            )
            for r in records
        ]

apply_schema()

Apply all constraints and indexes idempotently.

Safe to call on every startup; uses IF NOT EXISTS guards.

Source code in src/tng/repository/graph_repository.py

def apply_schema(self) -> None:
    """Apply all constraints and indexes idempotently.

    Safe to call on every startup; uses ``IF NOT EXISTS`` guards.
    """
    with self._driver.session(database=self._db) as session:
        for stmt in Q.SCHEMA_CONSTRAINTS + Q.SCHEMA_INDEXES:
            session.run(stmt)
    logger.info("Schema constraints and indexes verified.")

apply_transform(transform)

Dispatch and persist a transformation on a scene.

Routes to the appropriate axis-specific Cypher query. The old state node is detached (not deleted) and the new state node is created in a single managed transaction.

Parameters:

Name	Type	Description	Default
transform	Transform	Fully populated Transform domain object.	required

Returns:

Type	Description
Transform	The same transform (with produced_id set if applicable).

Raises:

Type	Description
ValueError	For unknown or unsupported axis values.

Source code in src/tng/repository/graph_repository.py

def apply_transform(self, transform: Transform) -> Transform:
    """Dispatch and persist a transformation on a scene.

    Routes to the appropriate axis-specific Cypher query.  The old
    state node is detached (not deleted) and the new state node is
    created in a single managed transaction.

    :param transform: Fully populated Transform domain object.
    :returns: The same transform (with ``produced_id`` set if applicable).
    :raises ValueError: For unknown or unsupported axis values.
    """
    axis = transform.axis
    dispatch = {
        TransformAxis.POV: self._apply_pov,
        TransformAxis.MOOD: self._apply_mood,
        TransformAxis.GENRE: self._apply_genre,
        TransformAxis.CHRONOTOPE: self._apply_chronotope,
        TransformAxis.RELIABILITY: self._apply_reliability,
        TransformAxis.CODE_OVERLAY: self._apply_code_overlay,
    }
    handler = dispatch.get(axis)
    if handler is None:
        raise ValueError(f"Unknown transform axis: {axis}")
    return handler(transform)

archive_narrative(narrative_id)

Set a Narrative's status to archived.

Parameters:

Name	Type	Description	Default
narrative_id	str	Target narrative ID.	required

Returns:

Type	Description
bool	True if the narrative was found and archived.

Source code in src/tng/repository/graph_repository.py

def archive_narrative(self, narrative_id: str) -> bool:
    """Set a Narrative's status to archived.

    :param narrative_id: Target narrative ID.
    :returns: ``True`` if the narrative was found and archived.
    """

    def _write(tx: Any) -> Any:
        return tx.run(Q.ARCHIVE_NARRATIVE, id=narrative_id).single()

    with self._driver.session(database=self._db) as session:
        result = session.execute_write(_write)
    return result is not None

close()

Release driver connection pool resources.

Source code in src/tng/repository/graph_repository.py

def close(self) -> None:
    """Release driver connection pool resources."""
    self._driver.close()

get_atom_revisions(atom_id)

Return all AtomRevision nodes for an atom, oldest first.

Parameters:

Name	Type	Description	Default
atom_id	str	The target Atom's ID.	required

Returns:

Type	Description
list[dict[str, Any]]	List of revision dicts with id, text, revised_at, operator, and reason keys.

Source code in src/tng/repository/graph_repository.py

def get_atom_revisions(self, atom_id: str) -> list[dict[str, Any]]:
    """Return all AtomRevision nodes for an atom, oldest first.

    :param atom_id: The target Atom's ID.
    :returns: List of revision dicts with ``id``, ``text``, ``revised_at``,
        ``operator``, and ``reason`` keys.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_ATOM_REVISIONS, atom_id=atom_id, database_=self._db
    )
    return [
        {
            "id": rec["r"]["id"],
            "atom_id": rec["r"]["atom_id"],
            "text": rec["r"]["text"],
            "revised_at": rec["r"]["revised_at"],
            "operator": rec["r"].get("operator", "system"),
            "reason": rec["r"].get("reason", ""),
        }
        for rec in records
    ]

get_atoms_with_context(narrative_id)

Fetch atoms in surface order with their current scene context.

Used by the prose renderer.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative to render.	required

Returns:

Type	Description
list[dict[str, Any]]	Row dicts with atom text, scene metadata, and perspective/mood.

Source code in src/tng/repository/graph_repository.py

def get_atoms_with_context(self, narrative_id: str) -> list[dict[str, Any]]:
    """Fetch atoms in surface order with their current scene context.

    Used by the prose renderer.

    :param narrative_id: The narrative to render.
    :returns: Row dicts with atom text, scene metadata, and perspective/mood.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_ATOMS_WITH_CONTEXT,
        narrative_id=narrative_id,
        database_=self._db,
    )
    return [dict(r) for r in records]

get_event_relations(narrative_id)

Fetch all inter-event causal and temporal relationships.

Returns CAUSES, ENABLES, PREVENTS, and PRECEDES edges between events within the narrative's scenes. Used by the GraphML renderer to draw and tension-score causal graph structure.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative to query.	required

Returns:

Type	Description
list[EventRelation]	List of EventRelation objects.

Source code in src/tng/repository/graph_repository.py

def get_event_relations(self, narrative_id: str) -> list[EventRelation]:
    """Fetch all inter-event causal and temporal relationships.

    Returns ``CAUSES``, ``ENABLES``, ``PREVENTS``, and ``PRECEDES``
    edges between events within the narrative's scenes.  Used by the
    GraphML renderer to draw and tension-score causal graph structure.

    :param narrative_id: The narrative to query.
    :returns: List of ``EventRelation`` objects.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_EVENT_RELATIONS,
        narrative_id=narrative_id,
        database_=self._db,
    )
    return [
        EventRelation(
            source_id=r["source_id"],
            target_id=r["target_id"],
            relation_type=r["relation_type"],
        )
        for r in records
    ]

get_graph_state(narrative_id)

Return a complete in-memory snapshot of a narrative's graph state.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative to snapshot.	required

Returns:

Type	Description
GraphState | None	A GraphState or None if the narrative doesn't exist.

Source code in src/tng/repository/graph_repository.py

def get_graph_state(self, narrative_id: str) -> GraphState | None:
    """Return a complete in-memory snapshot of a narrative's graph state.

    :param narrative_id: The narrative to snapshot.
    :returns: A ``GraphState`` or ``None`` if the narrative doesn't exist.
    """
    narrative = self.get_narrative(narrative_id)
    if narrative is None:
        return None
    transforms_raw = []
    for scene in narrative.scenes:
        transforms_raw.extend(self.get_transform_history(scene.id))

    transforms = [
        Transform(
            id=r["id"],
            axis=TransformAxis(r["axis"]),
            operator=r.get("operator", "system"),
            applied_at=_parse_dt(r.get("applied_at")),
            parameters=json.loads(r.get("parameters") or "{}"),
            scene_id=r.get("scene_id", ""),
            produced_id=r.get("produced_id", ""),
        )
        for r in transforms_raw
    ]
    event_relations = self.get_event_relations(narrative_id)
    return GraphState(
        narrative=narrative,
        transforms=transforms,
        event_relations=event_relations,
    )

get_narrative(narrative_id)

Retrieve a Narrative by ID with all nested scenes and atoms.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative's unique identifier.	required

Returns:

Type	Description
Narrative | None	A populated Narrative or None if not found.

Source code in src/tng/repository/graph_repository.py

def get_narrative(self, narrative_id: str) -> Narrative | None:
    """Retrieve a Narrative by ID with all nested scenes and atoms.

    :param narrative_id: The narrative's unique identifier.
    :returns: A populated ``Narrative`` or ``None`` if not found.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_NARRATIVE, id=narrative_id, database_=self._db
    )
    if not records:
        return None
    node = records[0]["n"]
    narrative = Narrative(
        id=node["id"],
        title=node["title"],
        status=NarrativeStatus(node.get("status", "draft")),
        source_ref=node.get("source_ref", ""),
        created_at=_parse_dt(node.get("created_at")),
    )
    narrative.scenes = self._get_scenes(narrative_id)
    return narrative

get_pattern(pattern_id)

Retrieve a single Pattern template by ID.

Parameters:

Name	Type	Description	Default
pattern_id	str	The pattern's unique identifier.	required

Returns:

Type	Description
Pattern | None	A Pattern or None if not found.

Source code in src/tng/repository/graph_repository.py

def get_pattern(self, pattern_id: str) -> Pattern | None:
    """Retrieve a single Pattern template by ID.

    :param pattern_id: The pattern's unique identifier.
    :returns: A ``Pattern`` or ``None`` if not found.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_PATTERN, id=pattern_id, database_=self._db
    )
    if not records:
        return None
    node = records[0]["p"]
    return Pattern(
        id=node["id"],
        name=node["name"],
        family=node["family"],
        description=node.get("description", ""),
    )

get_scene_ids(narrative_id)

Return scene IDs for a narrative in sequence order.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative's unique identifier.	required

Returns:

Type	Description
list[str]	List of scene ID strings, ordered by sequence.

Source code in src/tng/repository/graph_repository.py

def get_scene_ids(self, narrative_id: str) -> list[str]:
    """Return scene IDs for a narrative in sequence order.

    :param narrative_id: The narrative's unique identifier.
    :returns: List of scene ID strings, ordered by sequence.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_SCENE_IDS_FOR_NARRATIVE,
        narrative_id=narrative_id,
        database_=self._db,
    )
    return [rec["scene_id"] for rec in records]

get_transform(transform_id)

Retrieve a Transform audit record by ID.

Parameters:

Name	Type	Description	Default
transform_id	str	The transform's unique identifier.	required

Returns:

Type	Description
dict[str, Any] | None	A dict with transform details or None if not found.

Source code in src/tng/repository/graph_repository.py

def get_transform(self, transform_id: str) -> dict[str, Any] | None:
    """Retrieve a Transform audit record by ID.

    :param transform_id: The transform's unique identifier.
    :returns: A dict with transform details or ``None`` if not found.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_TRANSFORM, id=transform_id, database_=self._db
    )
    if not records:
        return None
    r = records[0]
    return {
        "id": transform_id,
        "scene_id": r.get("scene_id"),
        "produced_type": r.get("produced_type"),
        "produced_id": r.get("produced_id"),
    }

get_transform_history(scene_id)

Return the ordered transformation history for a scene.

Parameters:

Name	Type	Description	Default
scene_id	str	The scene to query.	required

Returns:

Type	Description
list[dict[str, Any]]	List of transform dicts ordered by applied_at ASC.

Source code in src/tng/repository/graph_repository.py

def get_transform_history(self, scene_id: str) -> list[dict[str, Any]]:
    """Return the ordered transformation history for a scene.

    :param scene_id: The scene to query.
    :returns: List of transform dicts ordered by applied_at ASC.
    """
    records, _, _ = self._driver.execute_query(
        Q.GET_TRANSFORM_HISTORY, scene_id=scene_id, database_=self._db
    )
    return [dict(r) for r in records]

list_pattern_instances(narrative_id)

List all PatternInstances for a narrative with context.

Parameters:

Name	Type	Description	Default
narrative_id	str	The narrative to query.	required

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with instance, pattern, and scene_id.

Source code in src/tng/repository/graph_repository.py

def list_pattern_instances(self, narrative_id: str) -> list[dict[str, Any]]:
    """List all PatternInstances for a narrative with context.

    :param narrative_id: The narrative to query.
    :returns: List of dicts with instance, pattern, and scene_id.
    """
    records, _, _ = self._driver.execute_query(
        Q.LIST_PATTERN_INSTANCES,
        narrative_id=narrative_id,
        database_=self._db,
    )
    return [
        {
            "instance_id": r["pi"]["id"],
            "slot": r["pi"]["slot"],
            "confidence": r["pi"]["confidence"],
            "pattern_id": r["p"]["id"],
            "pattern_name": r["p"]["name"],
            "pattern_family": r["p"]["family"],
            "scene_id": r["scene_id"],
        }
        for r in records
    ]

list_patterns(family=None)

List pattern templates, optionally filtered by family.

Parameters:

Name	Type	Description	Default
family	str | None	If provided, only return patterns of this family.	None

Returns:

Type	Description
list[Pattern]	List of matching patterns.

Source code in src/tng/repository/graph_repository.py

def list_patterns(self, family: str | None = None) -> list[Pattern]:
    """List pattern templates, optionally filtered by family.

    :param family: If provided, only return patterns of this family.
    :returns: List of matching patterns.
    """
    records, _, _ = self._driver.execute_query(
        Q.LIST_PATTERNS, family=family, database_=self._db
    )
    return [
        Pattern(
            id=r["p"]["id"],
            name=r["p"]["name"],
            family=r["p"]["family"],
            description=r["p"].get("description", ""),
        )
        for r in records
    ]

ping()

Return True when Neo4j is reachable and the database responds.

Returns:

Type	Description
bool	True on success.

Raises:

Type	Description
Exception	On Neo4j connectivity failure — callers decide how to handle it.

Source code in src/tng/repository/graph_repository.py

def ping(self) -> bool:
    """Return True when Neo4j is reachable and the database responds.

    :returns: ``True`` on success.
    :raises Exception: On Neo4j connectivity failure — callers decide how to
        handle it.
    """
    result = self._driver.execute_query(
        Q.HEALTH_PING, database_=self._db
    )
    return bool(result.records)

revise_atom(atom_id, revision_id, text, revised_at, operator, reason)

Create an AtomRevision node and re-point CURRENT_REVISION.

The old CURRENT_REVISION edge is detached (not deleted) and a SUPERSEDES edge is added from the new revision to the old one.

Parameters:

Name	Type	Description	Default
atom_id	str	The target Atom's ID.	required
revision_id	str	Pre-generated ID for the new AtomRevision node.	required
text	str	Revised prose text.	required
revised_at	datetime	UTC timestamp.	required
operator	str	Identifier of the requesting user/system.	required
reason	str	Optional reason for the change.	required

Returns:

Type	Description
bool	True if the atom was found and revised.

Source code in src/tng/repository/graph_repository.py

def revise_atom(
    self,
    atom_id: str,
    revision_id: str,
    text: str,
    revised_at: datetime,
    operator: str,
    reason: str,
) -> bool:
    """Create an AtomRevision node and re-point CURRENT_REVISION.

    The old CURRENT_REVISION edge is detached (not deleted) and a
    SUPERSEDES edge is added from the new revision to the old one.

    :param atom_id: The target Atom's ID.
    :param revision_id: Pre-generated ID for the new AtomRevision node.
    :param text: Revised prose text.
    :param revised_at: UTC timestamp.
    :param operator: Identifier of the requesting user/system.
    :param reason: Optional reason for the change.
    :returns: ``True`` if the atom was found and revised.
    """

    def _write(tx: Any) -> Any:
        return tx.run(
            Q.CREATE_ATOM_REVISION,
            atom_id=atom_id,
            revision_id=revision_id,
            text=text,
            revised_at=_dt_str(revised_at),
            operator=operator,
            reason=reason,
        ).single()

    with self._driver.session(database=self._db) as session:
        result = session.execute_write(_write)
    return result is not None

save_narrative(narrative)

Persist or update a Narrative node.

Parameters:

Name	Type	Description	Default
narrative	Narrative	The domain Narrative to persist.	required

Returns:

Type	Description
Narrative	The same narrative (unchanged — the graph write is idempotent via MERGE).

Source code in src/tng/repository/graph_repository.py

def save_narrative(self, narrative: Narrative) -> Narrative:
    """Persist or update a Narrative node.

    :param narrative: The domain Narrative to persist.
    :returns: The same narrative (unchanged — the graph write is
        idempotent via MERGE).
    """

    def _write(tx: Any) -> None:
        tx.run(
            Q.MERGE_NARRATIVE,
            id=narrative.id,
            title=narrative.title,
            status=narrative.status.value,
            source_ref=narrative.source_ref,
            created_at=_dt_str(narrative.created_at),
        )

    with self._driver.session(database=self._db) as session:
        session.execute_write(_write)
    logger.debug("Saved narrative %s", narrative.id)
    return narrative

save_pattern(pattern)

Persist a Pattern template.

Parameters:

Name	Type	Description	Default
pattern	Pattern	The pattern to save.	required

Returns:

Type	Description
Pattern	The same pattern (unchanged).

Source code in src/tng/repository/graph_repository.py

def save_pattern(self, pattern: Pattern) -> Pattern:
    """Persist a Pattern template.

    :param pattern: The pattern to save.
    :returns: The same pattern (unchanged).
    """

    def _write(tx: Any) -> None:
        tx.run(
            Q.MERGE_PATTERN,
            id=pattern.id,
            name=pattern.name,
            family=pattern.family,
            description=pattern.description,
        )

    with self._driver.session(database=self._db) as session:
        session.execute_write(_write)
    return pattern

save_scene(scene, narrative_id)

Persist a Scene and link it to its parent Narrative.

Parameters:

Name	Type	Description	Default
scene	Scene	The domain Scene to persist.	required
narrative_id	str	The parent narrative's ID.	required

Returns:

Type	Description
Scene	The saved scene.

Source code in src/tng/repository/graph_repository.py

def save_scene(self, scene: Scene, narrative_id: str) -> Scene:
    """Persist a Scene and link it to its parent Narrative.

    :param scene: The domain Scene to persist.
    :param narrative_id: The parent narrative's ID.
    :returns: The saved scene.
    """

    def _write(tx: Any) -> None:
        tx.run(
            Q.MERGE_SCENE,
            id=scene.id,
            sequence=scene.sequence,
            summary=scene.summary,
            narrative_id=narrative_id,
        )
        for atom in scene.atoms:
            self._save_atom_tx(tx, atom, scene.id)
        for event in scene.events:
            self._save_event_tx(tx, event, scene.id)
        for instance in scene.pattern_instances:
            self._save_pattern_instance_tx(tx, instance, scene.id)

    with self._driver.session(database=self._db) as session:
        session.execute_write(_write)
    return scene

update_narrative_status(narrative_id, status)

Update the status property on a Narrative node.

Parameters:

Name	Type	Description	Default
narrative_id	str	Target narrative ID.	required
status	NarrativeStatus	New status value.	required

Source code in src/tng/repository/graph_repository.py

def update_narrative_status(
    self, narrative_id: str, status: NarrativeStatus
) -> None:
    """Update the status property on a Narrative node.

    :param narrative_id: Target narrative ID.
    :param status: New status value.
    """

    def _write(tx: Any) -> None:
        tx.run(
            Q.UPDATE_NARRATIVE_STATUS,
            id=narrative_id,
            status=status.value,
        )

    with self._driver.session(database=self._db) as session:
        session.execute_write(_write)

update_narrative_status_for_scene(scene_id, status)

Update the status of the Narrative that contains a given scene.

Parameters:

Name	Type	Description	Default
scene_id	str	ID of a scene whose parent narrative should be updated.	required
status	NarrativeStatus	New status value.	required

Source code in src/tng/repository/graph_repository.py

def update_narrative_status_for_scene(
    self, scene_id: str, status: NarrativeStatus
) -> None:
    """Update the status of the Narrative that contains a given scene.

    :param scene_id: ID of a scene whose parent narrative should be updated.
    :param status: New status value.
    """

    def _write(tx: Any) -> None:
        tx.run(
            """
            MATCH (n:Narrative)-[:HAS_SCENE]->(s:Scene {id: $scene_id})
            SET n.status = $status
            """,
            scene_id=scene_id,
            status=status.value,
        )

    with self._driver.session(database=self._db) as session:
        session.execute_write(_write)

create_driver(settings)

Create and return an authenticated Neo4j driver.

Parameters:

Name	Type	Description	Default
settings	Settings	Application settings containing Neo4j URI and credentials.	required

Returns:

Type	Description
Driver	A connected neo4j.Driver instance.

Source code in src/tng/repository/graph_repository.py

def create_driver(settings: Settings) -> Driver:
    """Create and return an authenticated Neo4j driver.

    :param settings: Application settings containing Neo4j URI and credentials.
    :returns: A connected ``neo4j.Driver`` instance.
    """
    return GraphDatabase.driver(
        settings.neo4j_uri,
        auth=(settings.neo4j_user, settings.neo4j_password),
    )