"""Work items in the state store (architecture L1, L7; Development Plan M1.1).""" import io from pathlib import Path from typing import Any import pytest from nisse_dev.canonical import AssistantText, UserMessage from nisse_dev.control import ( AWAIT_ARCH_SIGNOFF, PR_OPEN, AskHuman, Orchestrator, SubmitPlan, ) from nisse_dev.oplog import OpLog from nisse_dev.state import IDLE, StateStore from nisse_dev.transport import InMemoryTransport def new_store(path: Path) -> StateStore: return StateStore(path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) # --- the store --------------------------------------------------------------- def test_a_transcript_reads_only_its_own_work_item(tmp_path: Path) -> None: store = new_store(tmp_path) store.transcript("a", "operator").append(UserMessage(text="for a", source="m")) store.transcript("b", "operator").append(UserMessage(text="for b", source="m")) store.transcript(None, "operator").append(UserMessage(text="idle", source="m")) assert [e.event.text for e in store.transcript("a", "planner").read()] == ["for a"] assert [e.event.text for e in store.transcript(None, "planner").read()] == ["idle"] def test_every_event_carries_its_work_item_and_actor(tmp_path: Path) -> None: store = new_store(tmp_path) log = store.transcript("a", "executor") log.append(AssistantText(text="working", model="m")) log.as_actor("helper").append(AssistantText(text="helping", model="f")) assert [(e.work_item, e.actor) for e in store.all_events()] == [ ("a", "executor"), ("a", "helper"), ] def test_an_unknown_actor_is_refused(tmp_path: Path) -> None: with pytest.raises(ValueError, match="unknown actor"): new_store(tmp_path).transcript("a", "someone") def test_intents_are_owned_by_their_work_item(tmp_path: Path) -> None: store = new_store(tmp_path) store.transcript("a", "helper").begin_intent("delegation", {}, cause="t") [intent] = store.pending_intents() assert intent.work_item == "a" def test_work_items_and_the_active_one_survive_a_restart(tmp_path: Path) -> None: store = new_store(tmp_path) store.open_work_item("a", title="demo", scope="demo", position=IDLE, cause="t") store.set_active("a", cause="t") store.update_work_item("a", position="PLANNING_ARCH", data={"x": 1}, cause="t") store.close() store = new_store(tmp_path) item = store.active_item() assert item is not None assert (item.id, item.position, item.data, item.status) == ( "a", "PLANNING_ARCH", {"x": 1}, "open", ) def test_closing_records_the_outcome_and_going_idle_clears_active( tmp_path: Path, ) -> None: store = new_store(tmp_path) store.open_work_item("a", title="demo", position=IDLE, cause="t") store.set_active("a", cause="t") store.close_work_item("a", outcome="completed", cause="t") store.set_active(None, cause="t") assert store.active_item() is None item = store.work_item("a") assert item is not None and (item.status, item.outcome) == ("closed", "completed") with pytest.raises(KeyError): store.close_work_item("a", outcome="abandoned", cause="t") # already closed def test_messages_sent_while_idle_are_carried_with_their_origin( tmp_path: Path, ) -> None: store = new_store(tmp_path) store.set_active(None, cause="boot") idle = store.transcript(None, "operator") first = idle.append(UserMessage(text="add a truncate helper", source="m:1")) store.open_work_item("a", title="demo", position=IDLE, cause="t") assert store.carry_idle_messages("a") == 1 [carried] = store.transcript("a", "planner").read() assert carried.event.text == "add a truncate helper" assert f"#{first}" in carried.event.source # where it came from assert len(store.all_events()) == 2 # a copy; the original is untouched # --- the orchestrator ---------------------------------------------------------- class RecordingPlanner: """Records what each planning turn could read.""" def __init__(self) -> None: self.seen: list[list[str]] = [] def plan( self, transcript: Any, gate: str, project: str, feedback: str | None = None, answer: str | None = None, ) -> SubmitPlan | AskHuman: self.seen.append( [ e.event.text for e in transcript.read() if isinstance(e.event, UserMessage) ] ) return SubmitPlan(f"{gate} plan for {project}") def build(tmp_path: Path) -> tuple[StateStore, InMemoryTransport, RecordingPlanner]: store = new_store(tmp_path) transport = InMemoryTransport() planner = RecordingPlanner() Orchestrator(store, transport, planner, operators={"don"}).start() return store, transport, planner def test_a_second_task_never_sees_the_first(tmp_path: Path) -> None: # The second live run's planner read every earlier task's events (#78). store, transport, planner = build(tmp_path) for text in ("build the first thing", "use first", "approve", "approve"): transport.receive("don", text) transport.receive("don", "start milestone 1") assert store.get_state().state == PR_OPEN transport.receive("don", "project complete") transport.receive("don", "build the second thing") transport.receive("don", "use second") assert store.get_state().state == AWAIT_ARCH_SIGNOFF second_task_view = planner.seen[-1] assert "build the second thing" in second_task_view assert not any("first" in text for text in second_task_view) def test_each_task_is_its_own_work_item_with_its_outcome(tmp_path: Path) -> None: store, transport, _ = build(tmp_path) for text in ("use first", "approve", "approve", "start milestone 1"): transport.receive("don", text) transport.receive("don", "project complete") transport.receive("don", "use second") transport.receive("don", "abandon") outcomes = [(item.title, item.outcome) for item in store.work_items()] assert outcomes == [("first", "completed"), ("second", "abandoned")] assert store.active_item() is None def test_the_task_message_before_use_reaches_the_planner(tmp_path: Path) -> None: # Run 2's flow: the task, then `use`. It must still be planned from. _, transport, planner = build(tmp_path) transport.receive("don", "add a truncate helper") transport.receive("don", "use demo") assert "add a truncate helper" in planner.seen[0]