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/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 79c8981a0e..12518c2c05 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 ( @@ -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() @@ -495,9 +491,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 +537,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 +552,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 +561,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 +575,37 @@ 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) + 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), + (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], 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), + (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 +618,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 +638,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 +666,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 +684,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 +732,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 +740,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 +761,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 +1039,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() == () 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}