dongarandClaude Opus 5.5
3fba958958
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m27s
A model no longer reads another role's turns as its own (#78). The log stays one record per work item; what a role is shown is projected from it by Transcript.view() (views.py), which the planner, executor and helper now build their context from: - own events unchanged; - operator messages labelled with their channel, e.g. [operator, Matrix] or [operator, comment on PR #165]; - orchestrator facts (a new `fact` event) as [orchestrator] statements; - the executor sees a helper's handback as [helper report], not its turns; - a helper sees the approved plan and the executor's milestone work as one-line narration; - the planner sees no other role's work at all. The orchestrator now writes facts at the points the models must know about: work item opened, plan approved or rejected, milestone started, finished or stopped, published (with PR number and URL) or not, review comment answered, next milestone, task completed or abandoned. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
153 lines
3.6 KiB
Python
153 lines
3.6 KiB
Python
"""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 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
|
|
| 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,
|
|
"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)
|