diff --git a/src/agent_comms/behaviour_patterns.py b/src/agent_comms/behaviour_patterns.py index 6abd717c28..9c765becc7 100644 --- a/src/agent_comms/behaviour_patterns.py +++ b/src/agent_comms/behaviour_patterns.py @@ -19,7 +19,7 @@ from .declared_family import DeclaredFamily from .goals import OrderCheck from .standing_rules import BoardDisagrees, PlanIdle, StandingRule -from .todos import PlainText, StorageStrikeTodo, Todo, TodoStore, VoidStrikeTodo +from .todos import PlainText, Todo, TodoStore, VoidStrikeTodo POSTMORTEM = "https://github.com/OpenHCSDev/toad/blob/main/docs/agents/CODEX-POSTMORTEM.md" COMMANDER_SKILL = "../../src/agent_comms/skills/comms-commander/SKILL.md" @@ -253,13 +253,9 @@ class UndeclaredAudiencePattern(BehaviourPattern): class RepeatedStrikePattern(BehaviourPattern): signature = ( - "A second storage strike with the same actor, owner and reason as a live strike, or a" - " send-back strike on an order already struck since it was sent." - ) - checks = ( - CallableCheck(StorageStrikeTodo.repeats), - CallableCheck(Todo.struck_on_order), + "A send-back strike on an order already struck since it was sent." ) + checks = (CallableCheck(Todo.struck_on_order),) sources = (ProblemFix("P184", "7af14a520"),) diff --git a/src/agent_comms/commander/tool_watch.py b/src/agent_comms/commander/tool_watch.py index 9c4e533495..a1621628c1 100644 --- a/src/agent_comms/commander/tool_watch.py +++ b/src/agent_comms/commander/tool_watch.py @@ -7,7 +7,6 @@ from ..pi_payloads import PiMessage, TextContent from ..tool_declaration import one_line from ..registration import Registration -from ..todos import StorageStrikeTodo def thread_names(registry: Registration) -> dict[str, str]: """Session file -> thread name, from the registry snapshot's threads.""" @@ -16,18 +15,13 @@ def thread_names(registry: Registration) -> dict[str, str]: def enforce_storage(registry: Registration, thread: str, call: CodingCall) -> None: - """Core judges its units and records each storage violation once.""" + """Explain a refused storage operation without punishing its unit.""" if thread not in order.units() or thread not in registry: return try: call.decide_storage(registry, thread) except SelectedToolDenied as violation: print(f'VIOLATION storage {thread}: {violation}', flush=True) - for todo in order.held_by(thread)[:1]: - order.strike(thread, todo.id, - f'{todo.id} broke the storage rule at a tool call: {violation}; ' - 'storage waste is banned outright', 'tool_watch at tool call', - kind=StorageStrikeTodo) def emit(turns, thread, final=''): diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 19c45babcb..83ca430dc6 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1708,23 +1708,8 @@ def apply(self, current: Todo) -> Todo: current.require_assignment(self.previous) _text(self.reason, "Strike reason", 2000) StrikeReason(self.reason) - if any(self.repeats(recorded) for recorded in current.live_strikes(self.previous.owner)): - return current return replace(current, strikes=(*current.strikes, self), revision=current.revision + 1) - def repeats(self, recorded: StrikeTodo) -> bool: - return False - - -@dataclass(frozen=True) -class StorageStrikeTodo(StrikeTodo): - """Repeated reports of one storage violation retain its recorded strike.""" - - def repeats(self, recorded: StrikeTodo) -> bool: - return (self.actor == recorded.actor and self.previous.owner == recorded.previous.owner - and self.reason == recorded.reason) - - @dataclass(frozen=True) class VerdictAudience: """The threads a send is for; a send reaching many threads wakes no thread outside it.""" diff --git a/tests/test_commander_tool_watch.py b/tests/test_commander_tool_watch.py index 58094d6500..181dad357c 100644 --- a/tests/test_commander_tool_watch.py +++ b/tests/test_commander_tool_watch.py @@ -9,7 +9,6 @@ from agent_comms.native_tool_call import CodingCall from agent_comms.native_tool_call import SelectedToolDenied -from agent_comms.todos import StorageStrikeTodo from test_commander_problem_orders import create @@ -110,7 +109,7 @@ def counted(cls, arguments, *, call_id): assert turns['worker'] == ['REFUSED write: Missing required fields for WriteTool: content', 'write b.py: body'] -def test_storage_violation_is_struck_once_across_watch_restart(tool_watch, commander, monkeypatch): +def test_storage_violation_keeps_orders_and_strikes_across_watch_restart(tool_watch, commander, monkeypatch, capsys): comms, _ = commander before = create(commander, 'P1') announcements = [] @@ -126,7 +125,7 @@ def deny(self, thread, effects): restarted = importlib.reload(tool_watch) restarted.enforce_storage(comms.registry, 'worker', call) after = comms.goals.todos.get('P1') - assert after.revision == before.revision + 1 - assert len(after.live_strikes('worker')) == 1 - assert isinstance(after.live_strikes('worker')[0], StorageStrikeTodo) - assert len(announcements) == 1 + assert after == before + assert after.live_strikes('worker') == () + assert announcements == [] + assert capsys.readouterr().out.count('VIOLATION storage worker:') == 3 diff --git a/tests/verify_storage_refusals.py b/tests/verify_storage_refusals.py new file mode 100644 index 0000000000..4146d63dd2 --- /dev/null +++ b/tests/verify_storage_refusals.py @@ -0,0 +1,71 @@ +"""Carry a copy of the actual board and re-read one known-refused native tool call. + +No tool executes: tool-watch reads the original native frame through its production +parser and storage guard. Only the copied stores can receive board writes. +""" + +from __future__ import annotations + +import collections +import io +import json +import shutil +import sys +from contextlib import closing, redirect_stdout +from dataclasses import replace +from pathlib import Path + +from agent_comms.commander import order, tool_watch +from agent_comms.comms import wire +from agent_comms.coordination_database import CoordinationStore +from agent_comms.registry_store import RegistryStore +from agent_comms.todos import TodoStore +from cutover.carry_storage_refusals import StorageRefusals + + +def copy_store(source: Path, destination: Path) -> None: + with source.open("rb") as stream: + sqlite = stream.read(16) == b"SQLite format 3\0" + if sqlite: + with closing(CoordinationStore.read_snapshot(source)) as reader: + with closing(CoordinationStore.writer(destination)) as writer: + reader.backup(writer) + else: + shutil.copy2(source, destination) + + +def main() -> None: + root = Path(sys.argv[1]) + root.mkdir(parents=True, exist_ok=False) + original = order.COMMS + unit = "pi-cmd-tools" + session = original.registry.require(unit).session_file + assert session, "The registered native source must exist" + for source in (TodoStore.path_in(original.root), RegistryStore.at(original.root).path): + copy_store(source, root / source.name) + first = StorageRefusals.carry(root, apply=True) + assert StorageRefusals.carry(root, apply=True).changed == 0 + print(json.dumps(first.report("carry_storage_refusals", True)), flush=True) + copied = wire(root) + order.COMMANDER = replace(order.COMMANDER, comms=copied) + order.COMMS = copied + board = copied.goals.todos + before = board.for_commander(copied.registry.require(order.ME).incarnation) + output = io.StringIO() + # This command was refused, with a recorded result, during P265. It is read + # as watcher history only, never sent to a tool runner or provider again. + marker = b"p265-${monitor}-initial.log" + with Path(session).open("rb") as source, redirect_stdout(output): + for line in source: + if marker in line: + tool_watch.follow(copied.registry, collections.defaultdict(list), unit, line) + after = board.for_commander(copied.registry.require(order.ME).incarnation) + assert before == after, "Watching the refusal changed the copied board" + assert "VIOLATION storage pi-cmd-tools:" in output.getvalue(), output.getvalue() + print(output.getvalue().strip()) + print("copied board unchanged by refusal; all assignments and strike counts preserved") + print("original stores opened read-only; no provider input or tool command replayed") + + +if __name__ == "__main__": + main() diff --git a/tools/cutover/carry_storage_refusals.py b/tools/cutover/carry_storage_refusals.py new file mode 100644 index 0000000000..bf11aba630 --- /dev/null +++ b/tools/cutover/carry_storage_refusals.py @@ -0,0 +1,79 @@ +"""Keep old storage strikes on the board, but void every one still counting.""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import replace +from pathlib import Path +from typing import cast + +from agent_comms.field_codec import FieldCodec +from agent_comms.todos import StrikeTodo, Todo, TodoStore, VoidStrikeTodo + +from .carry import Carried, TodoRows + + +class StorageRefusals(TodoRows): + old_kind = "storage_strike_todo" + reason = "storage refusals refuse the command and never strike" + + @classmethod + def carry(cls, root: Path, *, apply: bool) -> Carried: + if not cls.store.path_in(root).exists(): + return Carried(0) + changed = voids = 0 + with cls.todo_rows(root, apply=apply) as connection: + db = cast(sqlite3.Connection, connection) + table = Todo.declared_name + plain_kind = FieldCodec.encode(StrikeTodo, type[StrikeTodo]) + names = db.execute( + f'SELECT id FROM "{table}" WHERE ' + "EXISTS(SELECT 1 FROM json_tree(strikes) WHERE key='kind' AND value=?) OR " + "EXISTS(SELECT 1 FROM json_tree(voided) WHERE key='kind' AND value=?)", + (cls.old_kind, cls.old_kind), + ).fetchall() + for (identity,) in names: + indices = db.execute( + f'SELECT j.key FROM "{table}" t,json_each(t.strikes) j ' + "WHERE t.id=? AND json_extract(j.value,'$.kind')=?", + (identity, cls.old_kind), + ).fetchall() + db.execute( + f'UPDATE "{table}" SET strikes=(SELECT json_group_array(json(part)) FROM ' + "(SELECT CASE WHEN json_extract(value,'$.kind')=? " + "THEN json_set(value,'$.kind',?) ELSE value END AS part " + "FROM json_each(strikes) ORDER BY key))," + "voided=(SELECT json_group_array(json(part)) FROM " + "(SELECT CASE WHEN json_extract(value,'$.strike.kind')=? " + "THEN json_set(value,'$.strike.kind',?) ELSE value END AS part " + "FROM json_each(voided) ORDER BY key)) WHERE id=?", + ( + cls.old_kind, + plain_kind, + cls.old_kind, + plain_kind, + identity, + ), + ) + current = TodoStore._current(db, identity) + for (index,) in indices: + strike = current.strikes[index] + if strike in current.live_strikes(): + # A carry records the owner's exact eight-word reason; + # it does not replay a command with a ten-word minimum. + taken_back = VoidStrikeTodo( + current.revision, current.creator_identity, strike, cls.reason + ) + current = replace( + current, + voided=(*current.voided, taken_back), + revision=current.revision + 1, + ) + voids += 1 + current.upsert(db) + changed += len(indices) + return cls.carried(root, changed, {"strikes": changed, "voids": voids}, apply=apply) + + +if __name__ == "__main__": + StorageRefusals.main()