P272: refuse storage commands without striking orders

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"),)
 
 

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=''):

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."""

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

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()

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()