Skip to content
Merged
10 changes: 4 additions & 6 deletions src/agent_comms/commander/judge.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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, ...]
Expand All @@ -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()}')

Expand Down Expand Up @@ -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)
Expand Down
17 changes: 11 additions & 6 deletions src/agent_comms/external_transcript.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)})'
Expand Down Expand Up @@ -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)
Expand Down
12 changes: 9 additions & 3 deletions src/agent_comms/native_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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),)


Expand Down
5 changes: 3 additions & 2 deletions src/agent_comms/tool_declaration.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -25,7 +26,7 @@ class ReplacementEdit:
@dataclass(frozen=True)
class EditCall:
id: str
at: float
at: EventTime
thread: str
path: str
hunks: tuple[ReplacementEdit, ...]
Expand Down Expand Up @@ -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:
Expand Down
4 changes: 3 additions & 1 deletion src/agent_comms/transcript_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
9 changes: 5 additions & 4 deletions src/agent_comms/turn_digests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
73 changes: 73 additions & 0 deletions tests/test_edit_evidence.py
Original file line number Diff line number Diff line change
@@ -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)
13 changes: 13 additions & 0 deletions tests/test_standing_rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
25 changes: 15 additions & 10 deletions tests/test_turn_digests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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']
Expand All @@ -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))
Expand Down Expand Up @@ -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:
Expand Down
Loading