P272: refuse storage operations without strikes (replaces #2037)

src/agent_comms/behaviour_patterns.py

@@ -18,7 +18,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"
@@ -356,14 +356,8 @@ class PlainTextPattern(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),
-    )
+    signature = "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

@@ -6,7 +6,6 @@
 from ..pi_payloads import PiMessage, StorageDenialToolDetails, TextContent
 from ..tool_declaration import one_line
 from ..registration import Registration
-from ..todos import StorageStrikeTodo
 from .order import Commander
 
 def thread_names(registry: Registration) -> dict[str, str]:
@@ -17,15 +16,10 @@ def thread_names(registry: Registration) -> dict[str, str]:
 
 def enforce_storage(ctx: Commander, thread: str, call_id: str,
                     denial: StorageDenialToolDetails) -> None:
-    """Report only the preflight refusal recorded in this call's native result."""
+    """Explain the actual preflight refusal without changing the problem or striking its unit."""
     if thread not in ctx.units() or thread not in ctx.comms.registry:
         return
-    print(f'VIOLATION storage {thread}: {denial.reason}', flush=True)
-    for todo in ctx.held_by(thread)[:1]:
-        ctx.strike(thread, todo.id,
-                   f'{todo.id} broke the storage rule at tool call {call_id}: {denial.reason}; '
-                   'storage waste is banned outright', 'tool_watch at tool call preflight',
-                   kind=StorageStrikeTodo)
+    print(f'VIOLATION storage {thread}: {denial.reason} (tool call {call_id})', flush=True)
 
 
 def emit(turns, thread, final=''):

src/agent_comms/storage_rules.py

@@ -211,7 +211,7 @@ def order_text(self, unit: str, pythons: Sequence[str]) -> str:
         """The rule as an order states it to ``unit``."""
         home = self.checkout_base / unit
         return (
-            f'Storage (enforced, automatic strike): your checkouts are {home}--P<n>, at most '
+            f'Storage (refused without a strike): your checkouts are {home}--P<n>, at most '
             f'{self.max_checkouts} besides your home checkout {home}, removed with git worktree '
             'remove in the same order that created them; your only scratch dir is '
             f'{self.scratch_base / unit}/, capped at {self.scratch_cap_gib:g} GB together with '

src/agent_comms/todos.py

@@ -2071,15 +2071,6 @@ 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, kw_only=True)
 class ObjectiveStrike(StrikeTodo, declared_name=None):
     """An objective failure recorded in the existing strike history, once per order and kind."""

tests/test_commander_tool_watch.py

@@ -9,7 +9,6 @@
 
 from agent_comms.native_tool_call import CodingCall
 from agent_comms.pi_payloads import AssistantMessage, StorageDenialToolDetails, ToolCallContent
-from agent_comms.todos import StorageStrikeTodo
 from test_commander_problem_orders import create
 
 
@@ -112,7 +111,8 @@ def counted(cls, arguments, *, call_id):
     assert turns['worker'] == ['REFUSED write: content: Missing required fields for WriteTool: content; fix: supply {"type": "string"}', 'write b.py: body']
 
 
-def test_storage_violation_is_struck_once_across_watch_restart(tool_watch, commander, commander_script, monkeypatch):
+def test_storage_violation_keeps_orders_and_strikes_across_watch_restart(
+        tool_watch, commander, commander_script, monkeypatch, capsys):
     comms, _ = commander
     before = create(commander, 'P1')
     announcements = []
@@ -133,7 +133,7 @@ def forbidden(*args):
     restarted = importlib.reload(tool_watch)
     restarted.follow(commander_script[0], turns, 'worker', line)
     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,78 @@
+"""Read a real refused native result on a board copy; never run its tool again.
+
+Usage: python tests/verify_storage_refusals.py ORIGINAL_ROOT NEW_COPY_ROOT UNIT
+Only the copied registry/Todo stores may be written. Native history is read-only.
+"""
+from __future__ import annotations
+
+import collections
+import io
+import json
+import shutil
+import sys
+from contextlib import closing, redirect_stdout
+from pathlib import Path
+
+from agent_comms.commander.config import CommanderConfig
+from agent_comms.commander.order import Commander
+from agent_comms.commander import tool_watch
+from agent_comms.coordination_database import CoordinationStore
+from agent_comms.native_entries import NativeEntry
+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:
+    original, root = Path(sys.argv[1]), Path(sys.argv[2])
+    unit = sys.argv[3]
+    config = CommanderConfig.read_root(original)
+    root.mkdir(parents=True, exist_ok=False)
+    for source in (TodoStore.path_in(original), RegistryStore.at(original).path):
+        copy_store(source, root / source.name)
+    first = StorageRefusals.carry(root, apply=True)
+    assert StorageRefusals.carry(root, apply=True).changed == 0
+    context = Commander(**config.model_dump(exclude={'root', 'state', 'board', 'narrative'}),
+                        root=root, state=root/'state', board=root/'fleet.md',
+                        narrative=root/'fleet-narrative.md')
+    assert unit in context.units()
+    session = context.comms.registry.require(unit).session_file
+    assert session
+    before = context.comms.goals.todos.for_commander(context.incarnation)
+    # Bound the saved-history read. Missing proof is a refusal, never a success.
+    with Path(session).open('rb') as source:
+        source.seek(0, 2)
+        size = source.tell()
+        source.seek(max(0, size - 4 * 1024 * 1024))
+        if source.tell():
+            source.readline()
+        lines = source.readlines()
+    frame = next((line for line in reversed(lines)
+                  if any(tuple(message.storage_refusals())
+                         for message in NativeEntry.read(line).messages)), None)
+    assert frame is not None, 'No typed storage refusal in the bounded native history tail'
+    output = io.StringIO()
+    with redirect_stdout(output):
+        for _ in range(3):
+            tool_watch.follow(context, collections.defaultdict(list), unit, frame)
+    after = context.comms.goals.todos.for_commander(context.incarnation)
+    assert before == after
+    assert output.getvalue().count(f'VIOLATION storage {unit}:') == 3
+    print(json.dumps(first.report('carry_storage_refusals', True)))
+    print(output.getvalue().strip())
+    print('Copied board unchanged across three reads; original stores/history read-only; no tool replay')
+
+
+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()