diff --git a/src/agent_comms/commander/judge.py b/src/agent_comms/commander/judge.py index 07affa6d45..2f975fdb19 100644 --- a/src/agent_comms/commander/judge.py +++ b/src/agent_comms/commander/judge.py @@ -32,7 +32,6 @@ BRANCH = re.compile(r'\bisinstance\(|\bhasattr\(|^\s*(if|elif|match|case)\b|\bif\b.*\belse\b') -@dataclass(frozen=True) class Hunk(ReplacementEdit): def branches_added(self) -> int: @@ -45,7 +44,6 @@ def rendered(self) -> str: + ''.join(f'+ {line}\n' for line in self.new_text.splitlines())) -@dataclass(frozen=True) class Edit(EditCall): """The Core result row, decoded at the commander's tool boundary.""" hunks: tuple[Hunk, ...] @@ -69,8 +67,7 @@ def added(self) -> str: return ''.join('+' + line + '\n' for hunk in self.hunks for line in hunk.new_text.splitlines()) def rendered(self) -> str: - stamp = time.strftime('%H:%M', time.localtime(self.at)) - return (f'--- {self.id} {stamp} {self.path}\n' + return (f'--- {self.id} {self.at.clock()} {self.path}\n' + '\n'.join(hunk.rendered() for hunk in self.hunks) + f'\n=== end {self.id} code {self.code()}') @@ -101,12 +98,13 @@ def roast(ctx, audience, thread, lines, text, fix, slop): def alarm(ctx): while True: - late = [edit for edit in Edit.unjudged(ctx.comms, ctx.commander) if time.time() - edit.at > 300] + late = [edit for edit in Edit.unjudged(ctx.comms, ctx.commander) + if edit.at.before(time.time() - 300)] if late: counts: dict[str, int] = {} for edit in late: counts[edit.thread] = counts.get(edit.thread, 0) + 1 - oldest = min(edit.at for edit in late) + oldest = min(edit.at.recorded_at("An overdue edit") for edit in late) print(f'UNJUDGED {len(late)} edits over 5 min (oldest {int((time.time() - oldest) / 60)} min): ' + ', '.join(f'{t} {n}' for t, n in counts.items()) + ' -> judge show', flush=True) time.sleep(60) diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index 0c9d70fd36..36495b72d2 100644 --- a/src/agent_comms/external_transcript.py +++ b/src/agent_comms/external_transcript.py @@ -23,6 +23,7 @@ from .recorded_time import EventTime, GeneratedTimestamp, UnrecordedEventTime from .native_transcript import RecordsCut from .thread_identity import ThreadIncarnation +from .tool_declaration import ToolDeclaration from .transcript_events import TranscriptEvent, TurnDigest, ToolStartTranscript from .transcript_sources import NoDraft, TranscriptDraft, TranscriptRecord, TranscriptSource from .typed_table import Column, ForeignKey, Index, RealStorage, StoreWriter, TypedRow, TypedStore, TypedTable @@ -212,7 +213,8 @@ def unjudged_query(cls, thread, todo): judged = tuple(verdict.tool_call_id for verdict in todo.verdicts) placeholders = lambda values: ",".join("?" for _ in values) since = todo.order.at.recorded_at("An order") - where = f'e.thread=? AND e.kind=? AND e.tool IN ({placeholders(tools)}) AND e.timestamp>=?' + where = (f'e.thread=? AND e.kind=? AND e.tool IN ({placeholders(tools)}) ' + 'AND (e.timestamp>=? OR e.timestamp IS NULL)') parameters = (thread, ToolStartTranscript.declared_name, *tools, since) if judged: where += f' AND e.call_id NOT IN ({placeholders(judged)})' @@ -364,14 +366,17 @@ def turns(self, bound: TurnBound) -> tuple[TurnDigest, ...]: def unjudged(self, todo): where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo) with closing(self.connection.execute( - 'SELECT event FROM (SELECT e.event, e.seq, e.ordinal, ' + 'SELECT timestamp,call_id,json_extract(event,\'$.tool\') FROM ' + '(SELECT e.event, e.timestamp, e.call_id, e.seq, e.ordinal, ' 'row_number() OVER (PARTITION BY e.call_id ORDER BY e.seq DESC,e.ordinal DESC) AS latest ' f'FROM "{ExternalEventRow.declared_name}" e WHERE {where}) ' 'WHERE latest=1 ORDER BY seq DESC,ordinal DESC', parameters)) as rows: - for encoded, in rows: - event = FieldCodec.decode(TranscriptEvent, FieldCodec.parse(encoded)) - yield from event.edit_calls(todo.held().owner, todo.id, - todo.order.at.recorded_at("An order")) + for timestamp, call_id, encoded in rows: + identity = FieldCodec.decode(str, call_id) + tool = FieldCodec.decode_bound(ToolDeclaration, FieldCodec.parse(encoded), + call_id=identity) + yield from tool.edit_calls(identity, GeneratedTimestamp.decode(timestamp), + todo.held().owner, todo.id) def unjudged_count(self, todo) -> int: where, parameters = ExternalEventRow.unjudged_query(self.transcript.lookup, todo) diff --git a/src/agent_comms/native_tools.py b/src/agent_comms/native_tools.py index e288825c84..d4dc9e90c4 100644 --- a/src/agent_comms/native_tools.py +++ b/src/agent_comms/native_tools.py @@ -11,9 +11,11 @@ from .envelope_claim_transitions import ExistingFileClaim, FileClaimPath, WritableFileClaim from .field_codec import FieldCodec from .native_tool_call import CodingCall, SelectedToolDenied +from .recorded_time import EventTime from .storage_rules import WriteStorageEffect, physical_path, shell_storage_effects from .tool_declaration import EditCall, ReplacementEdit, ToolDeclaration, one_line + class NativeToolDeclaration(ToolDeclaration): """A declared tool participating in the existing native owner check.""" @@ -63,8 +65,11 @@ def catalog(cls) -> list[dict[str, object]]: return [tool.schema() for tool in ToolDeclaration.members_with(cls)] @classmethod - def create_call(cls, call_id: str, arguments: dict[str, object]) -> CoordinationCall: - return CoordinationCall(call_id=call_id, declared_name=cls.declared_name, arguments=arguments) + def create_call( + cls, call_id: str, arguments: dict[str, object] + ) -> CoordinationCall: + return CoordinationCall( + call_id=call_id, declared_name=cls.declared_name, arguments=arguments) @classmethod def native_extension(cls, package: Path) -> Path: @@ -171,7 +176,8 @@ def _parse_resource(self): def storage_effects(self, cwd): return (WriteStorageEffect(physical_path(self.path, cwd)),) - def edit_calls(self, call_id, at, thread, problem): + def edit_calls(self, call_id: str, at: EventTime, thread: str, + problem: str) -> tuple[EditCall, ...]: return (EditCall(call_id, at, thread, self.path, self.replacements(), problem),) diff --git a/src/agent_comms/tool_declaration.py b/src/agent_comms/tool_declaration.py index cedcbea414..ea109bd1f3 100644 --- a/src/agent_comms/tool_declaration.py +++ b/src/agent_comms/tool_declaration.py @@ -10,6 +10,7 @@ from .declared_family import DeclaredFamily, UnknownFamilyMember from .field_codec import FieldCodec, MalformedRecord from .native_tool_call import SelectedToolDenied +from .recorded_time import EventTime def one_line(text: str, limit: int) -> str: @@ -25,7 +26,7 @@ class ReplacementEdit: @dataclass(frozen=True) class EditCall: id: str - at: float + at: EventTime thread: str path: str hunks: tuple[ReplacementEdit, ...] @@ -67,7 +68,7 @@ def record(cls, name: str, arguments: object, *, call_id: str) -> ToolDeclaratio except (MalformedRecord, SelectedToolDenied): return None - def edit_calls(self, call_id: str, at: float, thread: str, problem: str) -> tuple[EditCall, ...]: + def edit_calls(self, call_id: str, at: EventTime, thread: str, problem: str) -> tuple[EditCall, ...]: return () def reads(self) -> str: diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 79fd3dbeb0..ea82ca4184 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -14,6 +14,7 @@ from .field_codec import FieldCodec from .messages import Message from .message_reference import MessageReference +from .recorded_time import EventTime from .routing import MessageRoute, TurnRouting from .tool_results import SENT_MESSAGE_MIME, ToolDiff, ToolText from .transcript_merge import EventMerge, StreamingMerge @@ -349,7 +350,8 @@ class ToolStartTranscript(ToolTranscript): def edit_calls(self, thread: str, problem: str, since: float) -> tuple[tool_declaration.EditCall, ...]: if self.before(since) or self.tool is None: return () - return self.tool.edit_calls(self.tool_call_id, self.timestamp, thread, problem) + return self.tool.edit_calls( + self.tool_call_id, EventTime.capture(self.timestamp), thread, problem) raw_input: dict[str, JsonValue] = field(default_factory=dict) # as the CLI sent it, for history tool: tool_declaration.ToolDeclaration | None = field( diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 69ba2f14f7..7a22d32421 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -8,12 +8,12 @@ from pathlib import Path from typing import TYPE_CHECKING +from .external_transcript import ExternalTranscript from .thread_execution import ConversationPreparation from .thread_identity import ThreadIncarnation -from .external_transcript import ExternalTranscript -from .transcript_events import TurnDigest -from .tool_result import ToolResult from .tool_declaration import EditCall +from .tool_result import ToolResult +from .transcript_events import TurnDigest if TYPE_CHECKING: from .threads import Thread @@ -92,7 +92,8 @@ def read(self) -> Result: return self.thread.execution.prepare_conversation(self) def native(self) -> Result: - turns = self.thread.runtime.history_source(self.root, self.thread.incarnation).turns(self.bound) + source = self.thread.runtime.history_source(self.root, self.thread.incarnation) + turns = source.turns(self.bound) return self.from_turns(turns) def external(self) -> Result: diff --git a/tests/test_edit_evidence.py b/tests/test_edit_evidence.py new file mode 100644 index 0000000000..94e301d62d --- /dev/null +++ b/tests/test_edit_evidence.py @@ -0,0 +1,73 @@ +"""Edit clocks come from their native event or the original external SQL timestamp.""" + +import pytest + +from agent_comms.commander.judge import Edit +from agent_comms.external_transcript import ExternalEventRow, ExternalTranscript +from agent_comms.field_codec import FieldCodec, MalformedRecord +from agent_comms.recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime +from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.tool_calls import CliToolArguments, ToolCall +from agent_comms.turn_digests import UnjudgedEdits +from dataclasses import replace + + +def original_call(): + arguments = {'file_path': 'owner.py', 'old_string': 'before', 'new_string': 'after'} + return CliToolArguments.decode(ToolCall('source-call', 'Edit', arguments), arguments) + + +def assert_edit(edit, clock): + assert edit.id == 'source-call' + assert edit.problem == 'P215' + assert edit.hunks[0].old_text == 'before' + assert edit.hunks[0].new_text == 'after' + assert edit.at == clock + assert '- before' in edit.rendered() + assert '+ after' in edit.rendered() + assert clock.clock() in edit.rendered() + + +def test_native_edit_keeps_its_recorded_time_through_reply_and_presentation(): + timestamp = 1792430000.0 + start, _ = original_call().finished('done', True, 12, timestamp) + evidence = start.edit_calls('unit', 'P215', 1.0) + wire = FieldCodec.encode(UnjudgedEdits(evidence)) + presented, = FieldCodec.decode(tuple[Edit, ...], wire['edits']) + assert_edit(presented, RecordedEventTime(timestamp)) + assert presented.thread == 'unit' + assert wire['edits'][0]['at'] == FieldCodec.encode(presented.at, EventTime) + + +@pytest.mark.parametrize('unrecorded', [False, True]) +def test_external_edit_clock_survives_actual_unjudged_tool(commander, unrecorded): + from test_commander_problem_orders import create + + comms, _ = commander + todo = create(commander, 'P215') + holder = comms.registry.require(todo.held().owner) + comms.registry.register(replace(holder, execution=ExternalThreadExecution)) + timestamp = todo.order.at.recorded_at('An order') + 1.0 + start, end = original_call().finished('done', True, 12, timestamp) + transcript = ExternalTranscript(comms.root, holder.incarnation) + transcript.finish(turn_id='original-turn', item_id='source-call', events=(start, end)) + if unrecorded: + # Reproduce the old publisher's wire shape only in this test-owned database. + with transcript.writing() as connection: + connection.execute( + f'UPDATE "{ExternalEventRow.declared_name}" ' + "SET event=json_set(event, '$.timestamp', NULL) WHERE thread=?", + (transcript.lookup,)) + edit, = Edit.unjudged(comms, 'lead') + clock = UnrecordedEventTime() if unrecorded else RecordedEventTime(timestamp) + assert_edit(edit, clock) + assert edit.thread == holder.name + with transcript.reading() as records: + assert records.unjudged_count(todo) == 1 + + +def test_recorded_clock_requires_its_original_float(): + with pytest.raises(MalformedRecord): + FieldCodec.decode(EventTime, {'kind': 'recorded', 'at': None}) + with pytest.raises(ValueError, match='finite Unix timestamp'): + RecordedEventTime(None) diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index 7cec6206e0..bfcd5ba98f 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -48,6 +48,19 @@ def lines(comms, member, commander="lead"): return [found.line() for found in member.violations(comms, commander)] +def test_wheel_spin_reads_external_owner_and_requires_actual_recent_turns(standing, tmp_path): + from agent_comms.transcript_events import AssistantTranscript + + comms, _, rules = standing + for name in ('recent-cli', 'old-cli', 'empty-cli'): + comms.registry.declare(comms.registry.new_thread( + name, frozenset(), str(tmp_path), execution=ExternalThreadExecution)) + for name, timestamp in (('recent-cli', time.time()), ('old-cli', 1.0)): + comms.transcripts.external(name).finish(turn_id='original-turn', item_id='reply', events=( + AssistantTranscript('retained answer', position=1, timestamp=timestamp),)) + assert [thread.name for thread in rules.WheelSpin.idle(comms, 'lead')] == ['recent-cli'] + + @pytest.mark.parametrize("reported,alarms", [ (NoCommitProgress(), True), (MergedProgress("d" * 9), False), diff --git a/tests/test_turn_digests.py b/tests/test_turn_digests.py index 105ca45654..d874f64f71 100644 --- a/tests/test_turn_digests.py +++ b/tests/test_turn_digests.py @@ -37,9 +37,11 @@ def test_external_turns_preserve_interleaving_and_the_whole_selected_turn(tmp_pa comms = prepared(tmp_path) comms.registry.declare(comms.registry.new_thread('cli', frozenset(), str(tmp_path), execution=ExternalThreadExecution)) source = comms.transcripts.external('cli') - for turn, item, text in [('old', 'a', 'older'), ('one', 'b', 'first'), - ('two', 'c', 'other'), ('one', 'd', 'continued')]: - source.finish(turn_id=turn, item_id=item, events=(AssistantTranscript(text),)) + for at, (turn, item, text) in enumerate([ + ('old', 'a', 'older'), ('one', 'b', 'first'), + ('two', 'c', 'other'), ('one', 'd', 'continued')], start=1): + source.finish(turn_id=turn, item_id=item, + events=(AssistantTranscript(text, position=item, timestamp=float(at)),)) answer = endpoint(comms, 'cli', 2) assert [turn['turn_id'] for turn in answer['turns']] == ['one', 'two'] assert [event['text'] for event in answer['turns'][0]['events']] == ['first', 'continued'] @@ -51,17 +53,19 @@ def test_native_turns_keep_original_calls_results_and_input_identity(tmp_path): source = tmp_path / 'native.jsonl' rows = [ dict(type='session', id='source', version=3), - dict(type='message', id='u1', message=dict(role='user', content='old input')), - dict(type='message', id='a1', parentId='u1', message=dict( + dict(type='message', id='u1', timestamp='2026-10-16T12:00:00Z', + message=dict(role='user', content='old input')), + dict(type='message', id='a1', parentId='u1', timestamp='2026-10-16T12:00:01Z', message=dict( role='assistant', content=[dict(type='text', text='old answer')], stopReason='stop')), - dict(type='message', id='u2', parentId='a1', message=dict(role='user', content='read file')), - dict(type='message', id='call', parentId='u2', message=dict( + dict(type='message', id='u2', parentId='a1', timestamp='2026-10-16T12:00:02Z', + message=dict(role='user', content='read file')), + dict(type='message', id='call', parentId='u2', timestamp='2026-10-16T12:00:03Z', message=dict( role='assistant', content=[dict(type='toolCall', id='read-1', name='read', arguments=dict(path='x.py'))], stopReason='toolUse')), - dict(type='message', id='result', parentId='call', message=dict( + dict(type='message', id='result', parentId='call', timestamp='2026-10-16T12:00:04Z', message=dict( role='toolResult', toolCallId='read-1', toolName='read', isError=False, content=[dict(type='text', text='contents')])), - dict(type='message', id='a2', parentId='result', message=dict( + dict(type='message', id='a2', parentId='result', timestamp='2026-10-16T12:00:05Z', message=dict( role='assistant', content=[dict(type='text', text='done')], stopReason='stop')), ] source.write_text(''.join(json.dumps(row) + '\n' for row in rows)) @@ -112,7 +116,8 @@ async def publish(): WatchSource.modified_file(item) for item in source.source_paths(comms.root)))) as changes: async for _ in changes: turns = await asyncio.to_thread(comms.transcripts.turns, 'claude', LastTurns()) - if any(event == AssistantTranscript('done', timestamp=datetime.fromisoformat(final['timestamp']).timestamp()) + if any(event == AssistantTranscript('done', position=event.position, + timestamp=datetime.fromisoformat(final['timestamp']).timestamp()) for turn in turns.turns for event in turn.events): break finally: