Give each role its own view of the transcript (M1.2)

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

nisse-dev/src/nisse_dev/canonical.py

@@ -73,6 +73,15 @@ class DelegationCompleted:
     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
@@ -89,6 +98,7 @@ CanonicalEvent = (
     | GateOpened
     | DelegationStarted
     | DelegationCompleted
+    | Fact
     | Compaction
 )
 
@@ -101,6 +111,7 @@ EVENT_TYPES: dict[str, type] = {
     "gate_opened": GateOpened,
     "delegation_started": DelegationStarted,
     "delegation_completed": DelegationCompleted,
+    "fact": Fact,
     "compaction": Compaction,
 }
 

nisse-dev/src/nisse_dev/control.py

@@ -22,7 +22,7 @@ from dataclasses import dataclass
 from typing import Any, Protocol
 
 from nisse_dev.cancel import Cancel
-from nisse_dev.canonical import GateOpened, PlanSubmitted, UserMessage
+from nisse_dev.canonical import Fact, GateOpened, PlanSubmitted, UserMessage
 from nisse_dev.gates import (
     Approved,
     Gate,
@@ -457,6 +457,11 @@ class Orchestrator:
         item = self._store.active_item()
         return self._store.transcript(item.id if item else None, actor)
 
+    def _fact(self, text: str) -> None:
+        """Record what the models must know and cannot see for themselves:
+        every role is shown it, labelled as the orchestrator's (L6)."""
+        self._log("orchestrator").append(Fact(text=text))
+
     def _close_item(self, outcome: str, cause: str) -> None:
         item = self._store.active_item()
         if item is not None and item.status == "open":
@@ -601,6 +606,7 @@ class Orchestrator:
         self._store.set_active(task_id, cause=f"operator command: use {project}")
         # What the operator said before `use` is part of this task (D1).
         self._store.carry_idle_messages(task_id)
+        self._fact(f"work item opened: project {project}, task {task_id}")
         if self._board is not None:
             try:
                 self._data["board"] = {
@@ -738,6 +744,10 @@ class Orchestrator:
                 channel,
                 f"{name} rev-{rev} rejected by {result.approver}: {result.feedback}",
             )
+            self._fact(
+                f"{name} plan rev-{rev} rejected by {result.approver}; "
+                "their feedback is the operator's message above"
+            )
             self._data["gate"] = None
             back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING
             self._transition(back, cause=f"plan rejected ({name} rev-{rev})")
@@ -754,6 +764,7 @@ class Orchestrator:
             rev=f"rev-{rev}",
             channel=result.channel,
         )
+        self._fact(f"{name} plan rev-{rev} approved by {result.approver}")
         if name == "arch":
             self._data["gate"] = None
             self._transition(
@@ -774,6 +785,7 @@ class Orchestrator:
             )
 
     def _execute_milestone(self, milestone: int) -> None:
+        self._fact(f"milestone {milestone} started")
         if self._executor is None:
             # The stub completes at once, so it still leaves EXECUTING.
             self._transport.send_message(f"[stub] executing milestone {milestone}")
@@ -783,6 +795,7 @@ class Orchestrator:
         with self._long_work():
             outcome = self._executor.run(self._log("executor"), milestone)
         if outcome.status == "stopped":
+            self._fact(f"milestone {milestone} stopped by the operator")
             self._halt_for_stop(
                 what=f"milestone {milestone}",
                 note=outcome.note,
@@ -794,9 +807,11 @@ class Orchestrator:
             )
             return
         if outcome.status == "completed":
+            self._fact(f"milestone {milestone} finished: {outcome.note}")
             self._transport.send_message(f"milestone {milestone} done: {outcome.note}")
             self._publish_milestone(milestone)
             return
+        self._fact(f"milestone {milestone} {outcome.status}: {outcome.note}")
         self._await_input(
             cause=f"milestone {milestone} {outcome.status}",
             prompt=f"milestone {milestone} stopped ({outcome.status}): {outcome.note}",
@@ -860,8 +875,11 @@ class Orchestrator:
         ):
             return  # not taken: a watch attached later still sees it
         self._take_comment(board_state, comment)
+        issue = (board_state or {}).get("issue")
         self._log("operator").append(
-            UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
+            UserMessage(
+                text=comment.body, source=f"gitea:issue:{issue}:comment:{comment.id}"
+            )
         )
         self._apply_gate_reply(
             sender=comment.author,
@@ -916,7 +934,10 @@ class Orchestrator:
         # The comment enters the transcript, so the executor sees review
         # feedback as conversation rather than as an out-of-band instruction.
         self._log("operator").append(
-            UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
+            UserMessage(
+                text=comment.body,
+                source=f"gitea:pr:{pr.get('number')}:comment:{comment.id}",
+            )
         )
         try:
             with self._long_work():
@@ -959,6 +980,7 @@ class Orchestrator:
             if sha
             else "no code change was needed for that comment"
         )
+        self._fact(f"review comment {comment.id}: {note}")
         self._transport.send_message(note)
         if pr.get("number"):
             self._publisher.comment(int(pr["number"]), note)
@@ -973,6 +995,7 @@ class Orchestrator:
     def _publish_milestone(self, milestone: int) -> None:
         if self._publisher is None:
             # No PR to wait on, but PR_OPEN's commands still lead on.
+            self._fact(f"milestone {milestone} done; no publisher, so no PR")
             self._transition(PR_OPEN, cause=f"[stub] milestone {milestone} done")
             self._transport.send_message(
                 f"[stub] milestone {milestone} done; no publisher, so no PR — "
@@ -991,6 +1014,7 @@ class Orchestrator:
             # The intent log already records what was in flight, so boot
             # reconciliation can settle it against Gitea. Gate rather than
             # retry blindly into a possible duplicate.
+            self._fact(f"publishing milestone {milestone} failed: {exc}")
             self._await_input(
                 cause=f"publishing milestone {milestone} failed: {exc}",
                 prompt=f"could not publish milestone {milestone}: {exc}",
@@ -1000,6 +1024,7 @@ class Orchestrator:
         if result.pull_request is None:
             # Nothing left EXECUTING behind: the operator decides what next,
             # and answering returns to the still-approved plan.
+            self._fact(f"milestone {milestone} not published: {result.note}")
             self._await_input(
                 cause=f"milestone {milestone}: {result.note}",
                 prompt=(
@@ -1009,6 +1034,10 @@ class Orchestrator:
                 pending={"kind": "milestone_stopped", "milestone": milestone},
             )
             return
+        self._fact(
+            f"milestone {milestone} published: {result.note} "
+            f"(PR #{result.pull_request.number}, {result.pull_request.url})"
+        )
         self._unwatch("plan")  # the gate is behind us; the PR is the thread now
         self._data["pr"] = {
             "number": result.pull_request.number,
@@ -1036,6 +1065,7 @@ class Orchestrator:
         pr = self._data.get("pr") or {}
         project = self._project
         self.stop_watching()
+        self._fact(f"work item abandoned by {sender}")
         self._close_item("abandoned", f"operator: abandon ({sender})")
         self._data = {**_fresh_data()}
         self._project = None
@@ -1111,7 +1141,14 @@ class Orchestrator:
         command = command_form(text)
         if command == "next milestone":
             self._unwatch("review")
+            pr = self._data.get("pr") or {}
             self._data["milestone"] = int(self._data.get("milestone", 1)) + 1
+            self._fact(
+                f"PR #{pr.get('number')} left for review; planning milestone "
+                f"{self._data['milestone']}"
+                if pr.get("number")
+                else f"planning milestone {self._data['milestone']}"
+            )
             self._data["gate"] = None
             self._data["pr"] = None
             self._transition(MILESTONE_PLANNING, cause="operator: next milestone")
@@ -1119,6 +1156,7 @@ class Orchestrator:
         elif command == "project complete":
             board_state = self._data.get("board") or {}
             self.stop_watching()
+            self._fact(f"work item completed: {sender} said project complete")
             self._close_item("completed", "operator: project complete")
             self._data = {**_fresh_data()}
             self._project = None

nisse-dev/src/nisse_dev/delegation.py

@@ -33,6 +33,7 @@ from nisse_dev.canonical import (
     AssistantText,
     DelegationCompleted,
     DelegationStarted,
+    PlanSubmitted,
     StoredEvent,
     ToolCall,
     ToolResult,
@@ -328,6 +329,7 @@ class Delegation:
         touched: set[str] = set()
         did_work = False
         spent = 0
+        after = _plan_scope(transcript)
 
         for turn in range(self._caps.turns):
             if self._cancel.requested:
@@ -350,7 +352,7 @@ class Delegation:
                 )
             try:
                 produced = self._client.complete(
-                    transcript.read(),
+                    transcript.view(after=after),
                     tools=DELEGATE_TOOLS,
                     model=self._model,
                     system=system,
@@ -580,6 +582,16 @@ def delegation_verdict(
     return "resume"
 
 
+def _plan_scope(transcript: Transcript) -> int:
+    """Where the helper's view begins: at the approved plan the work is under,
+    so it sees the plan and this milestone, not earlier ones (L6)."""
+    latest = 0
+    for stored in transcript.read():
+        if isinstance(stored.event, PlanSubmitted):
+            latest = stored.id
+    return max(latest - 1, 0)
+
+
 def _completed_after_start(events: list[StoredEvent]) -> bool:
     """True when the last delegation event in the log is a completion."""
     for stored in reversed(events):

nisse-dev/src/nisse_dev/dump.py

@@ -47,6 +47,8 @@ def _event_line(kind: str, payload: dict[str, Any], full: bool) -> str:
         case "plan_submitted":
             plan = payload["plan"]
             return f"PLAN ({payload.get('model', '')}, {len(plan)} chars): {_clip(plan, full)}"
+        case "fact":
+            return f"FACT: {_clip(payload['text'], full)}"
         case "gate_opened":
             return f"GATE {payload.get('gate', '')}: {payload.get('reason', '')}"
         case _:

nisse-dev/src/nisse_dev/executor.py

@@ -128,7 +128,7 @@ class LLMExecutor:
             f"Milestone {milestone} is stuck. The local executor tried and its "
             f"commands failed {failures} times in a row.\n\n"
             f"Most recent failure:\n{last_failure[:2000]}\n\n"
-            "The approved plan and the full transcript are in the conversation "
+            "The approved plan and the milestone so far are in the conversation "
             "above. Finish this sub-task, staying inside the approved plan."
         )
         result = self._delegate.run(
@@ -214,7 +214,7 @@ class LLMExecutor:
                 return ExecutorOutcome("stopped", f"before turn {turn + 1}")
             try:
                 produced = self._client.complete(
-                    transcript.read(after=after),
+                    transcript.view(after=after),
                     tools=tools,
                     model=model,
                     system=system,

nisse-dev/src/nisse_dev/planner.py

@@ -151,7 +151,7 @@ class LLMPlanningAgent:
     def _turn(
         self, transcript: Transcript, model: str, system: str
     ) -> SubmitPlan | AskHuman | None:
-        events = transcript.read()
+        events = transcript.view()
         produced = self._client.complete(
             events, tools=PLANNER_TOOLS, model=model, system=system
         )

nisse-dev/src/nisse_dev/render.py

@@ -16,6 +16,7 @@ from nisse_dev.canonical import (
     Compaction,
     DelegationCompleted,
     DelegationStarted,
+    Fact,
     GateOpened,
     PlanSubmitted,
     StoredEvent,
@@ -121,6 +122,8 @@ class OpenAIChatRenderer:
                     "role": "assistant",
                     "content": f"[delegation handback]\n{handback}",
                 }
+            case Fact(text=text):
+                return {"role": "user", "content": f"[orchestrator] {text}"}
             case Compaction(summary=summary):
                 return {
                     "role": "user",

nisse-dev/src/nisse_dev/render_anthropic.py

@@ -23,6 +23,7 @@ from nisse_dev.canonical import (
     Compaction,
     DelegationCompleted,
     DelegationStarted,
+    Fact,
     GateOpened,
     PlanSubmitted,
     StoredEvent,
@@ -152,6 +153,8 @@ class AnthropicMessagesRenderer:
                         {"type": "text", "text": f"[delegation handback]\n{handback}"},
                     )
                 ]
+            case Fact(text=text):
+                return [("user", {"type": "text", "text": f"[orchestrator] {text}"})]
             case Compaction(summary=summary):
                 return [
                     (

nisse-dev/src/nisse_dev/state.py

@@ -28,6 +28,7 @@ from nisse_dev.canonical import (
     to_payload,
 )
 from nisse_dev.oplog import OpLog
+from nisse_dev.views import project
 
 CURRENT_SCHEMA_VERSION = 4
 
@@ -619,8 +620,14 @@ class Transcript:
         return self._store._append(event, self.work_item, self.actor)
 
     def read(self, after: int = 0) -> list[StoredEvent]:
+        """The log as recorded: every role's events, for the record."""
         return self._store._read(self.work_item, after)
 
+    def view(self, after: int = 0) -> list[StoredEvent]:
+        """The log as this actor is shown it: what a model's context is
+        built from (L6). See ``views``."""
+        return project(self.read(after), self.actor)
+
     def as_actor(self, actor: str) -> "Transcript":
         """The same work item, written by another role: e.g. the executor
         handing a sub-task to the helper."""

nisse-dev/src/nisse_dev/views.py

@@ -0,0 +1,129 @@
+"""Per-role views of a work item's transcript (architecture L6; Development
+Plan M1.2).
+
+The log records every role's events in one work item. A model must not read
+another role's turns as its own: in the second live run the planner, shown
+the executor's tool calls as assistant turns, took that work for its own
+(#78). So what a role is shown is projected here, before rendering:
+
+- its own events, unchanged;
+- what the operator said, as a user message labelled with where it came from;
+- the orchestrator's facts and gates, as labelled statements;
+- another model role's work only where it needs it, and only as a labelled
+  report or narration, never as a turn.
+
+The projection keeps each event's id, so cursors (``after``) and compaction
+ranges still line up with the log. Everything not the viewer's own becomes a
+``UserMessage``: from where the viewer sits, it is all input.
+"""
+
+from dataclasses import replace
+
+from nisse_dev.canonical import (
+    AssistantText,
+    DelegationCompleted,
+    Fact,
+    GateOpened,
+    PlanSubmitted,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+
+# How much of another role's tool result a narration line keeps.
+NARRATION_RESULT_CHARS = 300
+
+# The roles whose work the viewer is shown, and how: "report" keeps only the
+# handback, "narrate" keeps a line per step. Anything not listed is not shown.
+_SHOWN: dict[str, dict[str, str]] = {
+    "planner": {},
+    "driver": {"helper": "report"},
+    "executor": {"helper": "report"},
+    "helper": {"executor": "narrate", "driver": "narrate"},
+}
+
+
+def operator_label(source: str) -> str:
+    """Where an operator message came from, for its label."""
+    origin, _, idle = source.partition(" (sent while idle")
+    if origin.startswith("transport:"):
+        channel = "Matrix"
+    elif origin.startswith("gitea:pr:"):
+        number = origin.split(":")[2]
+        channel = f"comment on PR #{number}"
+    elif origin.startswith("gitea:issue:"):
+        number = origin.split(":")[2]
+        channel = f"comment on plan issue #{number}"
+    elif origin.startswith("gitea:"):
+        channel = "Gitea comment"
+    else:
+        channel = origin or "unknown"
+    if idle:
+        channel += ", sent while idle"
+    return f"operator, {channel}"
+
+
+def _said(stored: StoredEvent, label: str, text: str) -> StoredEvent:
+    return replace(
+        stored,
+        event=UserMessage(text=f"[{label}] {text}", source=f"view:{stored.actor}"),
+    )
+
+
+def _narration(call: ToolCall, result: ToolResult | None) -> str:
+    target = call.arguments.get("path") or call.arguments.get("command")
+    step = f"{call.tool}({str(target)[:80]})" if target else call.tool
+    if result is None:
+        return f"{step} → no result recorded"
+    outcome = "ok" if result.success else "failed"
+    text = result.result.strip()
+    if len(text) > NARRATION_RESULT_CHARS:
+        text = text[:NARRATION_RESULT_CHARS] + " …"
+    return f"{step} → {outcome}: {text}" if text else f"{step} → {outcome}"
+
+
+def project(events: list[StoredEvent], viewer: str) -> list[StoredEvent]:
+    """The events as ``viewer`` should be shown them, in log order."""
+    shown = _SHOWN.get(viewer)
+    if shown is None:
+        # A role with no view defined (the expert is stateless, the
+        # classifier sees only an action) reads the log as it is.
+        return list(events)
+    results = {
+        e.event.call_id: e.event for e in events if isinstance(e.event, ToolResult)
+    }
+    out: list[StoredEvent] = []
+    for stored in events:
+        event, actor = stored.event, stored.actor
+        if isinstance(event, PlanSubmitted):
+            # A record for the executor's scope and the dump, not a turn:
+            # the planner's own submit_plan call already carries the plan.
+            if viewer == "helper":
+                out.append(_said(stored, "approved plan", event.plan))
+            continue
+        if actor == viewer:
+            out.append(stored)
+        elif actor == "operator" and isinstance(event, UserMessage):
+            out.append(_said(stored, operator_label(event.source), event.text))
+        elif isinstance(event, Fact):
+            out.append(_said(stored, "orchestrator", event.text))
+        elif isinstance(event, GateOpened):
+            out.append(
+                _said(stored, "orchestrator", f"gate {event.gate}: {event.reason}")
+            )
+        elif shown.get(actor) == "report" and isinstance(event, DelegationCompleted):
+            out.append(
+                _said(
+                    stored,
+                    f"{actor} report, {event.exit_status}",
+                    event.handback,
+                )
+            )
+        elif shown.get(actor) == "narrate":
+            if isinstance(event, ToolCall):
+                line = _narration(event, results.get(event.call_id))
+                out.append(_said(stored, actor, line))
+            elif isinstance(event, AssistantText) and event.text.strip():
+                out.append(_said(stored, actor, event.text.strip()))
+    return out