Batch mD-1: P233 fixture, native order wording (#2125 #2117)

src/agent_comms/goal_actions.py

@@ -529,6 +529,7 @@ def change(self, goal: Goal, ctx: GoalActionContext) -> Goal:
 
     def request_ref(self, ctx: GoalActionContext) -> GoalRef:
         """Refuse invalid orders before reserving their Todo."""
+        self.order.require_wording()
         commander = ctx.goals.registry.require(self.commander)
         commander.goal.require().require_id(self.request).require_active().order.require_request()
         self.deadline.require_deadline()

src/agent_comms/todos.py

@@ -1458,6 +1458,11 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None):
     def checked(self) -> CheckedProblemOrder:
         return self
 
+    def require_wording(self) -> None:
+        """Check fresh order prose, never reinterpret stored history or shell commands."""
+        for text in (*self.edits, *self.banned, self.report, *self.run.instructions()):
+            DirectiveText.require(text)
+
     def story(self) -> tuple[StoryLine, ...]:
         return (StoryLine(self.at, f"Ordered {self.id}: {'; '.join(self.edits)}"),)
 
@@ -2023,6 +2028,7 @@ def apply(self, current: Todo) -> Todo:
         current.require_revision(self.expected_revision)
         self.actor.require_may_change(current)
         if self.order != current.order:
+            self.order.require_wording()
             self.order.require_writable()
         assigned = (current if current.assignment == proposed
                     else current.state.assign(current, proposed, self.traced))
@@ -2996,6 +3002,8 @@ def create(
         goal: GoalRef,
     ) -> Todo:
         identity, assignee = _participant(creator), _thread(owner)
+        DirectiveText.require(text)
+        order.require_wording()
         order.require_writable()
         proposed = Todo.ordered(
             todo_id, repo, text, identity, Assignment.of(assignee, identity, generation),

tests/test_commander_problem_orders.py

@@ -61,6 +61,60 @@ def create(commander, problem, command='printf 1', expected=1, deadline=1000, ow
     return comms.goals.todos.get(problem)
 
 
+@pytest.mark.parametrize('text', (
+    'Repair the store; thank you for fixing it',
+    'Repairing the store is optional',
+    'Can you repair the store',
+))
+def test_new_native_order_refuses_soft_prose_before_reservation(commander, text):
+    comms, caller = commander
+    original = comms.registry.require('worker').goal
+    checked = replace(order(comms.root, 'soft-order'), edits=(text,))
+    # Old orders still decode: policy belongs to new input, not stored history.
+    assert FieldCodec.decode(PendingProblemOrder, FieldCodec.encode(checked)) == checked
+    with pytest.raises(ValueError, match='Soft wording'):
+        CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction(
+            problem='Psoft', order=checked, request='request',
+            deadline=RecordedEventTime(1000), commander='lead',
+            repo='OpenHCSDev/agent-comms', text='Repair the store',
+        )).apply(comms, caller)
+    assert comms.goals.todos.list('OpenHCSDev/agent-comms') == ()
+    assert comms.registry.require('worker').goal == original
+
+
+@pytest.mark.parametrize('field', ('edits', 'banned', 'report', 'run'))
+def test_direct_assignment_uses_the_same_wording_family(commander, field):
+    comms, _ = commander
+    current = create(commander, 'P1')
+    text = 'Please repair the store'
+    value = text if field == 'report' else (text,)
+    if field == 'run':
+        value = ApplicationOrderRun('Run the actual application and please show the result')
+    checked = replace(order(comms.root, 'replacement'), **{field: value})
+    with pytest.raises(ValueError, match='Soft wording'):
+        comms.goals.todos.assign(
+            'P1', expected_revision=current.revision,
+            owner=comms.registry.require('worker'),
+            assigner=comms.registry.require('lead').incarnation,
+            generation=current.held().generation, goal=current.goal,
+            deadline=current.deadline, order=checked,
+        )
+    assert comms.goals.todos.get('P1') == current
+
+
+def test_direct_creation_checks_defect_prose_before_writing(commander):
+    comms, _ = commander
+    lead = comms.registry.require('lead')
+    with pytest.raises(ValueError, match='Soft wording'):
+        comms.goals.todos.create(
+            'Psoft', 'OpenHCSDev/agent-comms', 'Please repair the store',
+            creator=lead, owner=comms.registry.require('worker'), generation='ongoing',
+            deadline=RecordedEventTime(1000), order=order(comms.root, 'soft-order'),
+            goal=GoalRef(lead.name, lead.created_at, 'request'),
+        )
+    assert comms.goals.todos.list('OpenHCSDev/agent-comms') == ()
+
+
 def working(comms):
     """The working claim a delivery stands on: a screenshot that exists."""
     screenshot = Path(comms.root) / 'working.png'

tests/test_turn_digests.py

@@ -82,8 +82,9 @@ def test_native_turns_keep_original_calls_results_and_input_identity(tmp_path):
 
 
 def test_claude_file_transport_reaches_the_same_cli_and_tool_answer(tmp_path, monkeypatch):
+    from agent_comms.child_process import ProcessIdentity
     from agent_comms.cli_bridges.claude import ClaudeBridge
-    from agent_comms.cli_bridges.shared import Binding, Options, Watch
+    from agent_comms.cli_bridges.shared import Binding, Options, Serve, Watch
     from agent_comms.wire_watch import WatchSource, WireWatch
 
     home = tmp_path / 'home'
@@ -105,7 +106,10 @@ def test_claude_file_transport_reaches_the_same_cli_and_tool_answer(tmp_path, mo
     store = Binding.open(state)
     db = store.__enter__()
     binding = Binding(str(comms.root), 'claude', 'session')
-    bridge = ClaudeBridge(Options(Watch(), binding, state), binding, db)
+    with pytest.raises(ValueError, match='does not own a live transcript subscription'):
+        Watch().transcript_source()
+    operation = Serve(owner=ProcessIdentity.capture(os.getpid()), transcript=path)
+    bridge = ClaudeBridge(Options(operation, binding, state), binding, db)
 
     async def publish():
         task = asyncio.create_task(bridge.publish_transcript())