From 2994ffcfcc920d62d899b57b9ac9674c2e718f16 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 14:40:10 -0400 Subject: [PATCH 1/3] Read the numbered terminator Claude shows after an export's final LF Every read-file export ends in LF, and Claude's Read numbers the empty row after it. ReadResultObserved.frames decoded that row as a ninth foreign frame, so read_export's all-lines gate refused all eight real parts and 358869593c89 sat at emitted forever on the installed b685915d9. frames() now drops exactly one final empty row, after the consecutive numbering holds, and only when it is not the only row. Every other empty or foreign row stays a frame the unchanged gate refuses. The test fake shown() modelled the Read from splitlines(), which never has that row, so every read-file test checked a Read Claude never produces. It now numbers physical LF rows of real export text; whole reads carry the terminator, partial pages do not. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- .../cli_bridges/claude_protocol.py | 7 +- tests/test_monitor_delivery.py | 111 ++++++++++++------ 2 files changed, 83 insertions(+), 35 deletions(-) diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index 03c9ab3d72..9fb8cd53be 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -680,7 +680,10 @@ def frames(self) -> tuple[MonitorFrame, ...]: """Claude shows each file line as its number, a tab, then the line. The leading numbered block is the file; numbers that do not run consecutively make no line count. Records end only at a physical LF: a JSON line may hold U+2028, U+0085 or other - characters str.splitlines would split on.""" + characters str.splitlines would split on. A Read reaching the end of a file that ends + in LF also numbers the empty row after that LF: once the numbering holds, that one + final empty row is the tool's terminator, not a record. Any other row, empty or not, + stays a frame the all-lines gate judges.""" numbers: list[int] = [] lines: list[str] = [] for row in self.text.split('\n'): @@ -692,6 +695,8 @@ def frames(self) -> tuple[MonitorFrame, ...]: start = numbers[0] if numbers else 0 if numbers != list(range(start, start + len(numbers))): return () + if len(lines) > 1 and lines[-1] == '': + lines.pop() return tuple(ClaudeMonitorFrame.decode_line(line) for line in lines) async def confirmed_by(self, bridge: LineMonitorBridge) -> RuntimeUpdate: diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index 79c8981a0e..c4e04e26bf 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -29,7 +29,7 @@ from agent_comms.child_process import ProcessIdentity from agent_comms.cli_bridges.monitor import MonitorPeer, MonitorService from agent_comms.cli_bridges.claude_protocol import ( - ClaudeSession, ClaudeMonitorFrame, ForeignTaskNotification, + ClaudeSession, ClaudeMonitorFrame, ForeignTaskNotification, ReadResultObserved, ForeignAttachment, KnownTurn, ) from agent_comms.cli_bridges.shared import ( @@ -495,9 +495,10 @@ def read_records(path, text, label, error=False): return (*read_started(session, path, label), *read_returned(session, text, label, error)) -def shown(lines, start=1): - """Claude's Read output: each line's number, a tab, then the line.""" - return '\n'.join(f'{number}\t{line}' for number, line in enumerate(lines, start)) +def shown(text, start=1): + """Claude's Read output of physical text: each LF-separated row's number, a tab, then the + row. Text ending in LF, as a whole export does, shows the empty row after it numbered too.""" + return '\n'.join(f'{number}\t{row}' for number, row in enumerate(text.split('\n'), start)) async def observe(bridge, updates): @@ -540,7 +541,7 @@ async def test_a_body_that_fits_one_event_goes_inline_and_exports_nothing(bridge @pytest.mark.asyncio -@pytest.mark.parametrize('parts', (2, 23, 25, 46, 26)) +@pytest.mark.parametrize('parts', (2, 8, 23, 25, 46, 26)) async def test_a_larger_message_is_one_notice_and_one_whole_read_confirms_it(bridge, parts): """The saved slow orders' part counts, and the smallest message that no longer fits one event: one bounded notice before anything is confirmed, then the whole body from one @@ -555,7 +556,8 @@ async def test_a_larger_message_is_one_notice_and_one_whole_read_confirms_it(bri assert notice == MonitorReadFile.naming(bridge.exports, message.reference) export = Path(notice.path) assert stat.S_IMODE(export.stat().st_mode) == 0o600 - lines = export.read_text().splitlines() + text = export.read_text(newline='') + lines = text.split('\n')[:-1] assert ''.join(ClaudeMonitorFrame.decode_line(item).body for item in lines) == body # The notice frees its own reservation and proves nothing about the body. await observe(bridge, notification(line, record_id=transcript_record('notice')).updates( @@ -563,7 +565,7 @@ async def test_a_larger_message_is_one_notice_and_one_whole_read_confirms_it(bri waiting, = await bridge.attempts(message.message_id) assert waiting.outcome is EmittedOutcome assert waiting.monitor == ReceivingMonitor(parts) - await observe(bridge, read_records(export, shown(lines), 'whole-read')) + await observe(bridge, read_records(export, shown(text), 'whole-read')) confirmed, = await bridge.attempts(message.message_id) assert confirmed.outcome is DeliveredOutcome assert confirmed.monitor == ObservedMonitor(transcript_record('whole-read'), 1792168805.123) @@ -577,25 +579,36 @@ async def test_a_larger_message_is_one_notice_and_one_whole_read_confirms_it(bri async def test_only_visible_pages_of_the_exact_export_confirm(bridge): message, export = await exported(bridge, body_of(23)) _, other_export = await exported(bridge, body_of(23) + 'y') - lines = export.read_text().splitlines() + text = export.read_text(newline='') + lines = text.split('\n')[:-1] copied = export.parent.parent / 'copied.jsonl' shutil.copyfile(export, copied) - for path, text, label, error in ( - (export, shown(lines), 'failed-read', True), - (copied, shown(lines), 'foreign-path', False), - (export, shown(other_export.read_text().splitlines()), 'foreign-message', False), - (export, shown(lines[:3]) + '\n5\t' + lines[4], 'numbering-gap', False), - (export, shown(lines[:5] + [lines[5][:100]]), 'truncated-line', False), + whole = shown(text) + # Only the one numbered empty row after the final LF is the Read's terminator. + for path, read, label, error in ( + (export, whole, 'failed-read', True), + (copied, whole, 'foreign-path', False), + (export, shown(other_export.read_text(newline='')), 'foreign-message', False), + (export, shown('\n'.join(lines[:3])) + '\n5\t' + lines[4], 'numbering-gap', False), + (export, shown('\n'.join([*lines[:5], lines[5][:100]])), 'truncated-line', False), + (export, shown('\n'.join([*lines[:-1], lines[-1][:100]]) + '\n'), 'truncated-final', + False), + (export, shown(text.replace('\n', '\n\n', 1)), 'interior-empty', False), + (export, shown(text + '\n'), 'two-terminal-empties', False), + (export, whole.rpartition('\n')[0] + f'\n{len(lines) + 2}\t', 'terminal-gap', False), + (export, shown(text + 'junk'), 'foreign-terminal', False), ): - await observe(bridge, read_records(path, text, label, error)) + await observe(bridge, read_records(path, read, label, error)) pending, = await bridge.attempts(message.message_id) assert pending.outcome is EmittedOutcome, label assert await bridge.coordinator_read( lambda db: HarnessObservedPart.of(db, pending)) == (), label - # Pages read out of order and repeated accumulate until every original part matches. - for start, stop, label in ((12, 23, 'last-page'), (12, 23, 'repeated-page'), - (0, 12, 'first-page')): - await observe(bridge, read_records(export, shown(lines[start:stop], start + 1), label)) + # Pages read out of order and repeated accumulate until every original part matches; a + # page reaching the end shows the terminator, a page stopping before it does not. + for page, start, label in (('\n'.join(lines[12:]) + '\n', 12, 'last-page'), + ('\n'.join(lines[12:]), 12, 'repeated-page'), + ('\n'.join(lines[:12]), 0, 'first-page')): + await observe(bridge, read_records(export, shown(page, start + 1), label)) confirmed, = await bridge.attempts(message.message_id) assert confirmed.outcome is DeliveredOutcome assert confirmed.monitor == ObservedMonitor(transcript_record('first-page'), 1792168805.123) @@ -608,17 +621,17 @@ async def test_an_export_copy_at_another_sequence_confirms_nothing(bridge): """The same message id at another sequence is not the attempted reference, even when every byte and part matches: the export names its exact reference.""" message, export = await exported(bridge, body_of(23)) - lines = export.read_text().splitlines() + text = export.read_text(newline='') copy = Path(MonitorReadFile.naming(bridge.exports, MessageReference( message.seq + 1000, message.message_id)).path) shutil.copyfile(export, copy) - await observe(bridge, read_records(copy, shown(lines), 'wrong-sequence')) + await observe(bridge, read_records(copy, shown(text), 'wrong-sequence')) pending, = await bridge.attempts(message.message_id) assert pending.outcome is EmittedOutcome assert pending.monitor == ReceivingMonitor(23) assert await bridge.coordinator_read(lambda db: HarnessObservedPart.of(db, pending)) == () assert export.exists() and copy.exists() - await observe(bridge, read_records(export, shown(lines), 'exact-sequence')) + await observe(bridge, read_records(export, shown(text), 'exact-sequence')) confirmed, = await bridge.attempts(message.message_id) assert confirmed.outcome is DeliveredOutcome assert confirmed.monitor == ObservedMonitor(transcript_record('exact-sequence'), 1792168805.123) @@ -628,10 +641,10 @@ async def test_an_export_copy_at_another_sequence_confirms_nothing(bridge): @pytest.mark.asyncio async def test_a_wrong_body_in_the_export_never_confirms(bridge, capsys): message, export = await exported(bridge, body_of(23)) - lines = export.read_text().splitlines() - forged = lines[0].replace('xxxx', 'xxxy', 1) - assert forged != lines[0] - await observe(bridge, read_records(export, shown([forged, *lines[1:]]), 'wrong-body')) + text = export.read_text(newline='') + forged = text.replace('xxxx', 'xxxy', 1) + assert forged != text + await observe(bridge, read_records(export, shown(forged), 'wrong-body')) stored, = await bridge.attempts(message.message_id) assert stored.outcome is EmittedOutcome assert isinstance(stored.monitor, MismatchedMonitor) @@ -656,11 +669,12 @@ async def test_line_separators_inside_a_body_stay_inside_their_part(bridge, sepa message, export = await exported(bridge, body) lines = export.read_text(newline='').split('\n')[:-1] assert ''.join(ClaudeMonitorFrame.decode_line(line).body for line in lines) == body - # Two visible pages of the actual numbered Read output, the later page first. + # Two visible pages of the actual numbered Read output, the later page, reaching the end + # and its terminator, first. session = ClaudeSession() - for start, stop in ((10, len(lines)), (0, 10)): + for start, page in ((10, '\n'.join(lines[10:]) + '\n'), (0, '\n'.join(lines[:10]))): await observe(bridge, (*read_started(session, export, f'page-{start}'), - *read_returned(session, shown(lines[start:stop], start + 1), + *read_returned(session, shown(page, start + 1), f'page-{start}'))) confirmed, = await bridge.attempts(message.message_id) assert confirmed.outcome is DeliveredOutcome @@ -673,7 +687,7 @@ async def test_line_separators_inside_a_body_stay_inside_their_part(bridge, sepa @pytest.mark.parametrize('ending', ('succeeded', 'failed')) async def test_a_consumed_export_stays_until_no_other_read_of_it_is_open(bridge, ending): message, export = await exported(bridge, body_of(23)) - text = shown(export.read_text().splitlines()) + text = shown(export.read_text(newline='')) session = ClaudeSession() starts = (*read_started(session, export, 'first'), *read_started(session, export, 'second')) await observe(bridge, (*starts, *read_returned(session, text, 'first'))) @@ -721,7 +735,7 @@ async def test_an_export_leaf_symlink_refuses(bridge, tmp_path): async def test_an_export_refuses_other_bytes_at_its_path(bridge): message, export = await exported(bridge, body_of(23)) notice = MonitorReadFile.naming(bridge.exports, message.reference) - notice.export(tuple(export.read_text().splitlines())) + notice.export(tuple(export.read_text(newline='').split('\n')[:-1])) with pytest.raises(RelationViolationError): notice.export(('"not this message"',)) @@ -729,13 +743,13 @@ async def test_an_export_refuses_other_bytes_at_its_path(bridge): @pytest.mark.asyncio async def test_an_uncertain_export_stays_and_reopening_never_replays_it(bridge): message, export = await exported(bridge, body_of(23)) - lines = export.read_text().splitlines() + text = export.read_text(newline='') # The coordinator's declared wait expires on a held lock; the read stays unproved. path = CoordinationStore.path_in(bridge.comms.root) with closing(sqlite3.connect(path, isolation_level=None)) as held: held.execute('BEGIN EXCLUSIVE') try: - await observe(bridge, read_records(export, shown(lines), 'busy-read')) + await observe(bridge, read_records(export, shown(text), 'busy-read')) finally: held.execute('ROLLBACK') uncertain, = await bridge.attempts(message.message_id) @@ -750,7 +764,7 @@ async def test_an_uncertain_export_stays_and_reopening_never_replays_it(bridge): assert await reopened.resume_after() == BridgeStart(message.reference) assert await reopened.attempts(message.message_id) == (uncertain,) # The same complete visible Read, seen after reopening, still confirms it. - await observe(reopened, read_records(export, shown(lines), 'later-read')) + await observe(reopened, read_records(export, shown(text), 'later-read')) confirmed, = await reopened.attempts(message.message_id) assert confirmed.outcome is DeliveredOutcome assert not export.exists() @@ -1028,3 +1042,32 @@ def test_only_commanded_sessions_reply_to_their_delivered_senders(comms): output.end_turn(turn) assert comms.bus.pending_count("unit") == 0 assert comms.bus.pending_count("lead") == 1 + + +@pytest.mark.parametrize(('rows', 'start', 'frames'), ( + (('A', 'B', ''), 1, (MonitorPart, MonitorPart)), + (('A', 'B', ''), 7, (MonitorPart, MonitorPart)), + (('A', 'B'), 1, (MonitorPart, MonitorPart)), + (('A', '', 'B', ''), 1, (MonitorPart, ForeignMonitorFrame, MonitorPart)), + (('A', 'B', '', ''), 1, (MonitorPart, MonitorPart, ForeignMonitorFrame)), + (('A', 'junk'), 1, (MonitorPart, ForeignMonitorFrame)), + (('',), 1, (ForeignMonitorFrame,)), +)) +def test_only_the_final_numbered_empty_row_is_the_read_terminator(rows, start, frames): + """Claude numbers the empty row after a file's final LF; that one row is not a record, + and every other empty or foreign row stays a frame the all-lines gate refuses.""" + message = Message(sender='peer', target='viewer', type=InfoKind, body='x' * 30000, + timestamp=1., seq=1) + first, second, *_ = (json.dumps(part, ensure_ascii=False) + for part in ClaudeMonitorFrame(message).parts) + physical = dict(A=first, B=second) + text = shown('\n'.join(physical.get(row, row) for row in rows), start) + observed = ReadResultObserved('export', text, ObservedMonitor(transcript_record('read'), 1.), + frozenset()) + assert tuple(type(frame) for frame in observed.frames()) == frames + + +def test_a_gap_at_the_terminator_number_makes_no_line_count(): + observed = ReadResultObserved('export', '1\t"x"\n3\t', + ObservedMonitor(transcript_record('read'), 1.), frozenset()) + assert observed.frames() == () From fbc4b10c715535f760b1fc79d65ef99eb1c145ce Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 14:50:57 -0400 Subject: [PATCH 2/3] Truncate the truncated-final control for real lines[-1][:100] on a final part shorter than 100 characters kept the whole record, so the negative case fed a valid export and expected refusal. Drop the last character instead and assert the result differs from the line. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- tests/test_monitor_delivery.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index c4e04e26bf..7ea6043513 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -584,6 +584,8 @@ async def test_only_visible_pages_of_the_exact_export_confirm(bridge): copied = export.parent.parent / 'copied.jsonl' shutil.copyfile(export, copied) whole = shown(text) + truncated = lines[-1][:-1] + assert truncated != lines[-1] # Only the one numbered empty row after the final LF is the Read's terminator. for path, read, label, error in ( (export, whole, 'failed-read', True), @@ -591,8 +593,7 @@ async def test_only_visible_pages_of_the_exact_export_confirm(bridge): (export, shown(other_export.read_text(newline='')), 'foreign-message', False), (export, shown('\n'.join(lines[:3])) + '\n5\t' + lines[4], 'numbering-gap', False), (export, shown('\n'.join([*lines[:5], lines[5][:100]])), 'truncated-line', False), - (export, shown('\n'.join([*lines[:-1], lines[-1][:100]]) + '\n'), 'truncated-final', - False), + (export, shown('\n'.join([*lines[:-1], truncated]) + '\n'), 'truncated-final', False), (export, shown(text.replace('\n', '\n\n', 1)), 'interior-empty', False), (export, shown(text + '\n'), 'two-terminal-empties', False), (export, whole.rpartition('\n')[0] + f'\n{len(lines) + 2}\t', 'terminal-gap', False), From d4c1eaf0c507d55286ab4d80e089fbb5b61951f1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 16:32:20 -0400 Subject: [PATCH 3/3] Restore the async stored_attempts reader and send the voice once My busy-begin test defined a second module-level stored_attempts(db, bridge, message) after the async stored_attempts(bridge), so Python kept the later one and the five async callers called a three-argument sync function with one argument. The one held-connection caller now selects HarnessDelivery on the held SQLite connection directly; the shadowing helper is gone and the async reader is the module's only one. CheckedProblemOrder.work_instructions pasted the whole voice file into every rendered order although UnitInstructions.render already carries it in each unit's system prompt: about a third of order bytes were that copy. The copy and its import are deleted; the test asserting the copy is deleted and the voice joins the UnitInstructions carrier test. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dR85kyFKcc2tQAJLhK --- src/agent_comms/todos.py | 2 -- tests/test_monitor_delivery.py | 10 +++------- tests/test_order_claims.py | 6 ------ tests/test_unit_instructions.py | 5 +++-- 4 files changed, 6 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index f79f0d566b..ac7577a0e3 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -63,7 +63,6 @@ OrderCheckFailedError, ) from agent_comms.command import Command -from agent_comms.commander.package import CommanderPackage from agent_comms.commander.rule_file import RuleAudience, RuleFile from agent_comms.coordination_session import NativeCapture from agent_comms.declared_family import DeclaredFamily @@ -2678,7 +2677,6 @@ def work_instructions(self) -> tuple[str, ...]: *self.deletes.instructions(), *self.run.instructions(), *(f"{n}. {edit}" for n, edit in enumerate(self.edits, 1)), "Banned: " + "; ".join(self.banned), "Report: " + self.report, - CommanderPackage.voice().read_text(encoding="utf-8").strip(), *(audience.order_line(", ".join(rule.name for rule in rules)) for audience, rules in RuleAudience.grouped(RuleFile.packaged()))) diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index 7ea6043513..12518c2c05 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -214,12 +214,6 @@ async def begin_waits(capsys, message): seen += capsys.readouterr().err -def stored_attempts(db, bridge, message): - return tuple(HarnessDelivery.select( - db, bridge=type(bridge), recipient_lookup=bridge.recipient_lookup, - message_id=message.message_id)) - - @pytest.mark.asyncio async def test_busy_begin_holds_the_same_message_until_the_store_frees(bridge, capsys): """Real SQLite contention before the durable begin: the original is held with no attempt @@ -238,7 +232,9 @@ async def test_busy_begin_holds_the_same_message_until_the_store_frees(bridge, c await begin_waits(capsys, held_message) # The event loop still runs other work while the begin waits. assert not accepting.done() - assert stored_attempts(held, bridge, held_message) == () + assert tuple(HarnessDelivery.select( + held, bridge=type(bridge), recipient_lookup=bridge.recipient_lookup, + message_id=held_message.message_id)) == () with pytest.raises(TimeoutError): async with asyncio.timeout(.3): await reader.readline() diff --git a/tests/test_order_claims.py b/tests/test_order_claims.py index 5898efa036..93be66f41a 100644 --- a/tests/test_order_claims.py +++ b/tests/test_order_claims.py @@ -11,7 +11,6 @@ from agent_comms.change_report import FileLine from agent_comms.commander.order import OrderFile, main -from agent_comms.commander.package import CommanderPackage from agent_comms.commander.standing import DeliveryAwaitingJudgement, Scrutinize from agent_comms.standing_rules import OwedAnswer from agent_comms.errors import TodoConflict, TodoError, ToolRefusedError @@ -74,11 +73,6 @@ def test_delivery_report_refuses_a_flat_status_memo(): assert DeliveryReport(blunt).text == blunt -def test_work_instructions_carry_the_owner_voice(source_repository): - heading = CommanderPackage.voice().read_text(encoding="utf-8").splitlines()[0] - assert any(line.startswith(heading) for line in order(source_repository).work_instructions()) - - def test_claim_verdict_family_on_retained_copy(held_commander): ctx = held_commander comms = ctx.comms diff --git a/tests/test_unit_instructions.py b/tests/test_unit_instructions.py index 4e618ca9bf..d9ebf785af 100644 --- a/tests/test_unit_instructions.py +++ b/tests/test_unit_instructions.py @@ -1,4 +1,4 @@ -"""Every native unit's system prompt carries the indictment, every rule under its audience and the pattern index.""" +"""Every native unit's system prompt carries the indictment, the voice, every rule under its audience and the pattern index.""" import pytest @@ -12,7 +12,8 @@ def test_unit_prompt_carries_every_rule_under_its_audience(tmp_path): text = UnitInstructions.prepare(tmp_path / "UNIT_SYSTEM.md").read_text(encoding="utf-8") - for source in (OWNER_INSTRUCTIONS, CommanderPackage.indictment(), Slop.catalog() / "README.md"): + for source in (OWNER_INSTRUCTIONS, CommanderPackage.indictment(), CommanderPackage.voice(), + Slop.catalog() / "README.md"): assert source.read_text(encoding="utf-8").strip() in text, source grouped = dict(RuleAudience.grouped(RuleFile.packaged())) assert set(grouped) == {FleetAudience, CommanderAudience}