Skip to content
9 changes: 7 additions & 2 deletions src/agent_comms/child_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -1065,6 +1065,10 @@ def close(self) -> None:
os.close(fd)


class OrderCheckFailedError(ValueError):
"""A completed order check did not prove its expected result."""


class CheckPrinted(DeclaredFamily, affix="Printed"):
"""The count printed by a child, or the absence of an integer."""

Expand Down Expand Up @@ -1102,7 +1106,7 @@ def plus(self, amount: int) -> CheckPrinted:

def require(self, expected: object) -> None:
if self.count:
raise ValueError(f"The order check is {self.count} away from {expected}.")
raise OrderCheckFailedError(f"The order check is {self.count} away from {expected}.")

def project[T](self, counted: Callable[[int], T],
uncounted: Callable[[NoIntegerPrinted], T]) -> T:
Expand All @@ -1122,7 +1126,8 @@ def plus(self, amount: int) -> CheckPrinted:
return self

def require(self, expected: object) -> None:
raise ValueError(f"The order check cannot prove {expected}: it printed no integer.")
raise OrderCheckFailedError(
f"The order check cannot prove {expected}: it printed no integer.")

def project[T](self, counted: Callable[[int], T],
uncounted: Callable[[NoIntegerPrinted], T]) -> T:
Expand Down
3 changes: 0 additions & 3 deletions src/agent_comms/commander/order.py
Original file line number Diff line number Diff line change
Expand Up @@ -601,9 +601,6 @@ def done(self, pid, sha, screenshot='', seen='', repo=''):
claim = MergedClaim(delivery.delivered)
self.call(CommsOrderDoneTool(problem=pid,
order_id=todo.order.id, delivery=delivery, claims=(claim,)))
delivered_todo = self.comms.goals.todos.get(pid)
for late in delivered_todo.strikes[len(todo.strikes):]:
self.posted(unit, delivered_todo, late)
print(f'{unit}: order {todo.order.id} delivered as {delivery.delivered if not operation else sha}')

def first_calls(self, held, activity):
Expand Down
9 changes: 5 additions & 4 deletions src/agent_comms/commander/standing.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,17 +179,18 @@ def dispatches(self, comms: Comms, commander: str) -> Iterable[WorkDispatch]:

@classmethod
def after_stall(cls, comms: Comms) -> None:
"""Assign first: refused or unavailable replacements leave the old holder untouched."""
stalled = comms.goals.todos.stalled(comms.registry.storage_rule)
for commander in {todo.creator_identity.name for todo in stalled}:
mine = (todo for todo in stalled if todo.creator_identity.name == commander)
"""Assign eligible work; refused replacements preserve the observed row."""
available = comms.goals.todos.reassignable(comms.registry.storage_rule)
for commander in {todo.creator_identity.name for todo in available}:
mine = (todo for todo in available if todo.creator_identity.name == commander)
for rule in cls.paired(comms, commander, mine, StallWatcherIncarnation.watcher()):
rule.enforce(comms, commander)

@classmethod
def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None:
"""The worker borrows the root for this pass, never while the timer is idle."""
with opened_wire() as comms:
comms.goals.todos.advance_objective_failures(comms.goals)
cls.after_stall(comms)


Expand Down
10 changes: 10 additions & 0 deletions src/agent_comms/external_transcript.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,7 @@ class ExternalEventRow(TypedTable):

foreign_keys = (ForeignKey(("thread", "seq"), ExternalItemRow, ("thread", "seq"), on_delete="CASCADE"),)
indexes = (Index(("thread", "kind", "tool", "timestamp", "call_id")),
Index(("thread", "kind", "timestamp")),
Index(("thread", "kind", "seq", "ordinal")),)

@classmethod
Expand Down Expand Up @@ -404,6 +405,15 @@ def source_paths(cls, root: Path) -> tuple[Path, ...]:
database = cls.path_in(root)
return database, database.with_name(database.name + "-wal")

def tool_call_between(self, since: float, through: float) -> bool:
"""Original tool timestamps decide first-call timeliness, not the last watcher tick."""
with self.snapshot() as connection:
return bool(connection.execute(
f'SELECT EXISTS(SELECT 1 FROM "{ExternalEventRow.declared_name}" '
'WHERE thread=? AND kind=? AND timestamp>=? AND timestamp<=?)',
(self.lookup, FieldCodec.encode(ToolStartTranscript), since, through),
).fetchone()[0])

def last_tool_call(self) -> EventTime:
with self.snapshot() as connection:
timestamp = connection.execute(
Expand Down
4 changes: 3 additions & 1 deletion src/agent_comms/thread_identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,9 @@ def require_may_change(self, todo: Todo) -> None:
"""The Core watcher may advance stalled orders without a commander turn."""

def assigned(self, previous: Todo, assigned: Todo) -> Todo:
return previous.reassigned(assigned, self)
if previous.holder:
return previous.reassigned(assigned, self)
return assigned

@classmethod
def watcher(cls) -> StallWatcherIncarnation:
Expand Down
Loading
Loading