Give each role its own view of the transcript (M1.2) #88
No files matched your search
@@ -386,8 +386,13 @@ the caller treats it as a failure (L9).
|
||||
happened from these, not by inferring it.
|
||||
- **Each role gets its own view:**
|
||||
- the driver sees the conversation, the facts, and helpers' reports;
|
||||
- a helper (delegation, `ask_expert`) sees its brief and its own work;
|
||||
- no role sees another role's raw turns presented as its own.
|
||||
- a delegated helper sees its brief, the approved plan, the facts, its own work,
|
||||
and the driver's work in the same milestone as compact narration (each step
|
||||
and how it went);
|
||||
- `ask_expert` is stateless: the expert sees only the question;
|
||||
- no role sees another role's raw turns presented as its own. Whatever is not
|
||||
the viewer's own reaches it as input labelled with its origin
|
||||
(`[orchestrator]`, `[operator, Matrix]`, `[helper report]`).
|
||||
|
||||
**3. Scope memory** — outlives work items.
|
||||
- **For a repo:** the project's own documents (`ARCHITECTURE.md`, build plans,
|
||||
|
||||
@@ -121,7 +121,7 @@ into Phase 2's design instead of being fixed twice.
|
||||
- Acceptance: two work items back to back keep separate state and events; a
|
||||
restart keeps both; migration tests.
|
||||
|
||||
**M1.2 Transcript views.** *Needs discussion* (rendering).
|
||||
**M1.2 Transcript views.** *Needs discussion* (rendering). **Done.**
|
||||
- Per-role rendering (L6): a role sees only its work item, attributed entries, and
|
||||
orchestrator `fact` events; never another role's raw turns as its own (#78).
|
||||
- Acceptance: the second live run's confusion can't happen (a planning call after
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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 _:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 [
|
||||
(
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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
|
||||
@@ -1,7 +1,7 @@
|
||||
import io
|
||||
from pathlib import Path
|
||||
|
||||
from nisse_dev.canonical import PlanSubmitted, ToolCall, ToolResult, UserMessage
|
||||
from nisse_dev.canonical import Fact, PlanSubmitted, ToolCall, ToolResult, UserMessage
|
||||
from nisse_dev.dump import main, render
|
||||
from nisse_dev.oplog import OpLog
|
||||
from nisse_dev.state import StateStore
|
||||
@@ -55,6 +55,15 @@ def test_full_shows_long_texts_whole(tmp_path: Path) -> None:
|
||||
assert "x" * 500 in render(db, full=True)
|
||||
|
||||
|
||||
def test_facts_are_shown_as_the_orchestrators(tmp_path: Path) -> None:
|
||||
db = tmp_path / "state.db"
|
||||
store_with_a_run(db)
|
||||
store = StateStore(db, oplog=OpLog("nisse-dev", io.StringIO()))
|
||||
store.transcript("demo-1", "orchestrator").append(Fact(text="PR #3 opened"))
|
||||
store.close()
|
||||
assert "[demo-1/orchestrator] FACT: PR #3 opened" in render(db)
|
||||
|
||||
|
||||
def test_the_dump_never_writes_to_the_store(tmp_path: Path) -> None:
|
||||
db = tmp_path / "state.db"
|
||||
store_with_a_run(db)
|
||||
|
||||
@@ -170,8 +170,8 @@ def test_planner_sees_prior_transcript(store: StateStore) -> None:
|
||||
)
|
||||
client = ScriptedClient([[plan_call("plan")]])
|
||||
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
|
||||
seen = [e.event for e in client.calls[0]["events"]]
|
||||
assert UserMessage(text="use demo", source="transport:@don") in seen
|
||||
seen = [getattr(e.event, "text", None) for e in client.calls[0]["events"]]
|
||||
assert "[operator, Matrix] use demo" in seen
|
||||
|
||||
|
||||
# --- orchestrator integration (still no network) -------------------------
|
||||
@@ -202,6 +202,7 @@ def test_use_project_produces_llm_plan_and_transcript(tmp_path: Path) -> None:
|
||||
kinds = [type(e).__name__ for e in events]
|
||||
assert kinds == [
|
||||
"UserMessage",
|
||||
"Fact",
|
||||
"ToolCall",
|
||||
"ToolResult",
|
||||
"PlanSubmitted",
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
"""Per-role transcript views (architecture L6; Development Plan M1.2)."""
|
||||
|
||||
import io
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from nisse_dev.canonical import (
|
||||
AssistantText,
|
||||
CanonicalEvent,
|
||||
DelegationCompleted,
|
||||
DelegationStarted,
|
||||
Fact,
|
||||
PlanSubmitted,
|
||||
StoredEvent,
|
||||
ToolCall,
|
||||
ToolResult,
|
||||
UserMessage,
|
||||
)
|
||||
from nisse_dev.control import MILESTONE_PLANNING, PR_OPEN, Orchestrator
|
||||
from nisse_dev.executor import ExecutorOutcome
|
||||
from nisse_dev.gitea import Comment, PullRequest
|
||||
from nisse_dev.llm import RoleMap
|
||||
from nisse_dev.oplog import OpLog
|
||||
from nisse_dev.planner import LLMPlanningAgent
|
||||
from nisse_dev.publish import PublishResult
|
||||
from nisse_dev.render import OpenAIChatRenderer
|
||||
from nisse_dev.render_anthropic import AnthropicMessagesRenderer
|
||||
from nisse_dev.state import StateStore
|
||||
from nisse_dev.transport import InMemoryTransport
|
||||
|
||||
ROLES = RoleMap(executor="x", planner="y", helper="y", expert="y")
|
||||
|
||||
|
||||
def new_store(tmp_path: Path) -> StateStore:
|
||||
return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
|
||||
|
||||
|
||||
def texts(events: list[StoredEvent]) -> list[str]:
|
||||
return [getattr(e.event, "text", "") for e in events]
|
||||
|
||||
|
||||
# --- the orchestrator: run 2's planner, after a milestone ----------------------
|
||||
|
||||
|
||||
class PlanningClient:
|
||||
"""Always submits a plan, recording the events of each request."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.requests: list[list[StoredEvent]] = []
|
||||
|
||||
def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
|
||||
self.requests.append(events)
|
||||
n = len(self.requests)
|
||||
return [
|
||||
ToolCall(
|
||||
call_id=f"plan-{n}",
|
||||
tool="submit_plan",
|
||||
arguments={"content": f"plan {n}"},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
class WorkingExecutor:
|
||||
"""Leaves the trail a real executor run does: text, calls, results."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.views: list[list[StoredEvent]] = []
|
||||
|
||||
def run(self, transcript: Any, milestone: int) -> ExecutorOutcome:
|
||||
self.views.append(transcript.view())
|
||||
transcript.append(AssistantText(text="I'll write the helper", model="x"))
|
||||
call = ToolCall(
|
||||
call_id=f"run-{len(self.views)}",
|
||||
tool="run_command",
|
||||
arguments={"command": "pytest"},
|
||||
)
|
||||
transcript.append(call)
|
||||
transcript.append(
|
||||
ToolResult(call_id=call.call_id, result="1 passed", success=True)
|
||||
)
|
||||
return ExecutorOutcome("completed", "added truncate() with a test")
|
||||
|
||||
|
||||
class Publisher:
|
||||
def reconcilers(self) -> dict[str, Any]:
|
||||
return {}
|
||||
|
||||
def publish(self, **kwargs: Any) -> PublishResult:
|
||||
return PublishResult(
|
||||
PullRequest(number=165, state="open", head="h", url="https://g/pr/165"),
|
||||
"opened PR #165",
|
||||
)
|
||||
|
||||
def push_followup(self, **kwargs: Any) -> str | None:
|
||||
return None
|
||||
|
||||
def comment(self, index: int, body: str) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def run_one_milestone(
|
||||
tmp_path: Path,
|
||||
) -> tuple[Orchestrator, StateStore, InMemoryTransport, PlanningClient, Any]:
|
||||
store = new_store(tmp_path)
|
||||
transport = InMemoryTransport()
|
||||
client = PlanningClient()
|
||||
executor = WorkingExecutor()
|
||||
orch = Orchestrator(
|
||||
store,
|
||||
transport,
|
||||
LLMPlanningAgent(client, ROLES),
|
||||
operators={"don"},
|
||||
roles=ROLES,
|
||||
executor=executor,
|
||||
publisher=Publisher(),
|
||||
gitea_operators={"don"},
|
||||
)
|
||||
orch.start()
|
||||
for text in ("add a truncate helper", "use demo", "approve", "approve"):
|
||||
transport.receive("don", text)
|
||||
transport.receive("don", "start milestone 1")
|
||||
assert store.get_state().state == PR_OPEN
|
||||
return orch, store, transport, client, executor
|
||||
|
||||
|
||||
def test_the_planner_after_a_milestone_sees_facts_not_executor_turns(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
# Run 2: shown the executor's calls as its own turns, the planner took
|
||||
# milestone 1's work for its own and planned around it (#78).
|
||||
_, store, transport, client, _ = run_one_milestone(tmp_path)
|
||||
transport.receive("don", "next milestone")
|
||||
assert store.get_state().state != MILESTONE_PLANNING # it planned
|
||||
|
||||
events = client.requests[-1]
|
||||
body = OpenAIChatRenderer().render(events, tools=None, model="y")
|
||||
assistant = [m for m in body["messages"] if m["role"] == "assistant"]
|
||||
called = [c["function"]["name"] for m in assistant for c in m["tool_calls"]]
|
||||
assert set(called) == {"submit_plan"} # its own turns, and only those
|
||||
seen = "\n".join(texts(events))
|
||||
assert "I'll write the helper" not in seen
|
||||
assert "[operator, Matrix, sent while idle] add a truncate helper" in seen
|
||||
assert "[orchestrator] arch plan rev-1 approved by don" in seen
|
||||
assert "milestone 1 finished: added truncate() with a test" in seen
|
||||
assert "PR #165" in seen and "https://g/pr/165" in seen
|
||||
assert "planning milestone 2" in seen
|
||||
|
||||
|
||||
def test_the_executor_sees_the_milestone_start_and_the_comment_source(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
orch, _, _, _, executor = run_one_milestone(tmp_path)
|
||||
assert "[orchestrator] milestone 1 started" in texts(executor.views[0])
|
||||
|
||||
orch.handle_review_comment(
|
||||
Comment(id=7, author="don", body="add a docstring", created_at="t")
|
||||
)
|
||||
review_view = texts(executor.views[-1])
|
||||
assert "[operator, comment on PR #165] add a docstring" in review_view
|
||||
# Its own earlier turn is still its own, not relabelled.
|
||||
own = [e for e in executor.views[-1] if isinstance(e.event, AssistantText)]
|
||||
assert [e.event.text for e in own] == ["I'll write the helper"]
|
||||
|
||||
|
||||
def test_the_outcome_is_recorded_as_a_fact_when_the_task_closes(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
_, store, transport, _, _ = run_one_milestone(tmp_path)
|
||||
item = store.active_item()
|
||||
transport.receive("don", "project complete")
|
||||
facts = [
|
||||
e.event.text
|
||||
for e in store.transcript(item.id, "orchestrator").read()
|
||||
if isinstance(e.event, Fact)
|
||||
]
|
||||
assert facts[-1] == "work item completed: don said project complete"
|
||||
|
||||
|
||||
# --- the projection, per role ----------------------------------------------------
|
||||
|
||||
|
||||
def delegated_milestone(store: StateStore) -> None:
|
||||
"""A milestone in which the executor failed and handed off to a helper."""
|
||||
store.transcript("a", "planner").append(PlanSubmitted(plan="1. fix it", model="y"))
|
||||
executor = store.transcript("a", "executor")
|
||||
executor.append(
|
||||
ToolCall(call_id="e1", tool="run_command", arguments={"command": "make"})
|
||||
)
|
||||
executor.append(ToolResult(call_id="e1", result="error: missing ;", success=False))
|
||||
helper = executor.as_actor("helper")
|
||||
helper.append(
|
||||
DelegationStarted(brief="make fails", role="helper", model="y", caps={})
|
||||
)
|
||||
helper.append(AssistantText(text="looking at main.c", model="y"))
|
||||
helper.append(
|
||||
ToolCall(call_id="h1", tool="read_file", arguments={"path": "main.c"})
|
||||
)
|
||||
helper.append(ToolResult(call_id="h1", result="int main()", success=True))
|
||||
helper.append(
|
||||
DelegationCompleted(
|
||||
handback="fixed the missing semicolon",
|
||||
artifacts=["main.c"],
|
||||
tokens=0,
|
||||
duration_s=1.0,
|
||||
exit_status="completed",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_the_executor_sees_the_helpers_report_not_its_turns(tmp_path: Path) -> None:
|
||||
store = new_store(tmp_path)
|
||||
delegated_milestone(store)
|
||||
view = store.transcript("a", "executor").view()
|
||||
assert [type(e.event).__name__ for e in view] == [
|
||||
"ToolCall",
|
||||
"ToolResult",
|
||||
"UserMessage",
|
||||
]
|
||||
assert view[-1].event.text == (
|
||||
"[helper report, completed] fixed the missing semicolon"
|
||||
)
|
||||
|
||||
|
||||
def test_the_helper_sees_the_plan_and_the_executors_work_as_narration(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
store = new_store(tmp_path)
|
||||
delegated_milestone(store)
|
||||
view = store.transcript("a", "helper").view()
|
||||
said = texts(view)
|
||||
assert said[0] == "[approved plan] 1. fix it"
|
||||
assert said[1] == "[executor] run_command(make) → failed: error: missing ;"
|
||||
# Its own events are unchanged, tool calls and all.
|
||||
assert any(isinstance(e.event, ToolCall) and e.event.call_id == "h1" for e in view)
|
||||
assert not any(
|
||||
isinstance(e.event, ToolCall) and e.event.call_id == "e1" for e in view
|
||||
)
|
||||
|
||||
|
||||
def test_the_planner_sees_no_other_roles_work(tmp_path: Path) -> None:
|
||||
store = new_store(tmp_path)
|
||||
delegated_milestone(store)
|
||||
store.transcript("a", "orchestrator").append(Fact(text="milestone 1 finished"))
|
||||
assert texts(store.transcript("a", "planner").view()) == [
|
||||
"[orchestrator] milestone 1 finished"
|
||||
]
|
||||
|
||||
|
||||
def test_a_view_keeps_log_ids_so_cursors_still_work(tmp_path: Path) -> None:
|
||||
store = new_store(tmp_path)
|
||||
delegated_milestone(store)
|
||||
raw = store.transcript("a", "executor").read()
|
||||
view = store.transcript("a", "executor").view(after=raw[2].id)
|
||||
assert [e.id for e in view] == [raw[-1].id] # the helper's report
|
||||
|
||||
|
||||
# --- rendering ---------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_both_renderers_show_a_fact_as_a_labelled_user_message() -> None:
|
||||
events = [StoredEvent(id=1, ts="", event=Fact(text="PR #3 opened"))]
|
||||
openai = OpenAIChatRenderer().render(events, tools=None, model="m")
|
||||
assert openai["messages"] == [
|
||||
{"role": "user", "content": "[orchestrator] PR #3 opened"}
|
||||
]
|
||||
anthropic = AnthropicMessagesRenderer().render(events, tools=None, model="m")
|
||||
assert anthropic["messages"] == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": "[orchestrator] PR #3 opened"}],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_operator_messages_carry_their_channel(tmp_path: Path) -> None:
|
||||
store = new_store(tmp_path)
|
||||
operator = store.transcript("a", "operator")
|
||||
for source in ("transport:@don", "gitea:issue:9:comment:4", "gitea:comment:5"):
|
||||
operator.append(UserMessage(text="hi", source=source))
|
||||
assert texts(store.transcript("a", "planner").view()) == [
|
||||
"[operator, Matrix] hi",
|
||||
"[operator, comment on plan issue #9] hi",
|
||||
"[operator, Gitea comment] hi",
|
||||
]
|
||||
Reference in new issue
Block a user