nisse/nisse-core/src/nisse_core/canonical.py

"""Canonical transcript schema (architecture L7).

The durable record is this event log, never a provider-shaped ``messages``
array; provider request bodies are rendered from it at call time. Canonical
tool-call IDs are minted locally — provider IDs are mapped at the boundary and
never used as a join key.

Transcript appends are the conversation record itself, not state-machine
mutations, so they are deliberately outside the L12 state-change taxonomy.
"""

import uuid
from dataclasses import asdict, dataclass, fields
from typing import Any


@dataclass(frozen=True)
class UserMessage:
    text: str
    source: str  # e.g. a Matrix event ID


@dataclass(frozen=True)
class AssistantText:
    text: str
    model: str


@dataclass(frozen=True)
class ToolCall:
    call_id: str  # canonical, minted locally via mint_call_id()
    tool: str
    arguments: dict[str, Any]


@dataclass(frozen=True)
class ToolResult:
    call_id: str
    result: str
    success: bool


@dataclass(frozen=True)
class PlanSubmitted:
    plan: str
    model: str


@dataclass(frozen=True)
class GateOpened:
    gate: str
    reason: str


@dataclass(frozen=True)
class DelegationStarted:
    brief: str
    role: str
    model: str  # the alias asked for, e.g. "remote-architecture"
    caps: dict[str, Any]  # turns / tokens / wall-clock limits
    # What the alias resolved to at task start. An alias is free to change
    # models between tasks, so eval history (L9) keyed on the alias alone
    # would average results across whatever it happened to point at.
    backing_model: str = ""


@dataclass(frozen=True)
class DelegationCompleted:
    handback: str
    artifacts: list[str]
    tokens: int
    duration_s: float
    exit_status: str


@dataclass(frozen=True)
class Fact:
    """Something the orchestrator knows that the models must too (L6, L7):
    a plan approved, a milestone finished, a PR opened. Written by the
    orchestrator; every role sees it, as a statement rather than a turn."""

    text: str


@dataclass(frozen=True)
class ActionVerdict:
    """What the fast path made of a tool call before it ran (P4, L4): a
    record for the log, the operator and the classifier's evaluation set,
    never shown to a model."""

    call_id: str
    tool: str
    effect: str  # harmless | contained | unknown | dangerous | refused
    rule: str
    reason: str
    decision: str  # ran | refused | asked | allowed | classified
    # Who decided: the fast path's rules, or the classifier (L4), with the
    # model it ran on and exactly what it was shown (the evaluation set, L9).
    source: str = "fast_path"
    model: str = ""
    input: str = ""


@dataclass(frozen=True)
class Compaction:
    summary: str
    model: str
    replaces: list[int]  # [first_id, last_id] inclusive range replaced


CanonicalEvent = (
    UserMessage
    | AssistantText
    | ToolCall
    | ToolResult
    | PlanSubmitted
    | GateOpened
    | DelegationStarted
    | DelegationCompleted
    | Fact
    | ActionVerdict
    | Compaction
)

EVENT_TYPES: dict[str, type] = {
    "user_message": UserMessage,
    "assistant_text": AssistantText,
    "tool_call": ToolCall,
    "tool_result": ToolResult,
    "plan_submitted": PlanSubmitted,
    "gate_opened": GateOpened,
    "delegation_started": DelegationStarted,
    "delegation_completed": DelegationCompleted,
    "fact": Fact,
    "action_verdict": ActionVerdict,
    "compaction": Compaction,
}

_TYPE_NAMES = {cls: name for name, cls in EVENT_TYPES.items()}


@dataclass(frozen=True)
class StoredEvent:
    id: int
    ts: str
    event: CanonicalEvent
    # The work item it belongs to (None: agent-level, L6) and the role that
    # produced it (L7). Rendering uses both.
    work_item: str | None = None
    actor: str = ""


def mint_call_id() -> str:
    return f"call-{uuid.uuid4().hex}"


def event_type(event: CanonicalEvent) -> str:
    return _TYPE_NAMES[type(event)]


def to_payload(event: CanonicalEvent) -> dict[str, Any]:
    return asdict(event)


def from_payload(type_name: str, payload: dict[str, Any]) -> CanonicalEvent:
    cls = EVENT_TYPES.get(type_name)
    if cls is None:
        raise ValueError(f"unknown canonical event type: {type_name!r}")
    known = {f.name for f in fields(cls)}
    unknown = payload.keys() - known
    if unknown:
        raise ValueError(f"unknown fields for {type_name}: {sorted(unknown)}")
    return cls(**payload)