From 152ec6ef1c90ebc584bb1cefb210a59a9fe4fc51 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 20:57:10 -0400 Subject: [PATCH 1/8] P283: retain each order and observe problem work across owned checkouts --- src/agent_comms/storage_rules.py | 17 +-- src/agent_comms/todos.py | 195 +++++++++++++++++++++---------- src/agent_comms/work_branches.py | 4 +- 3 files changed, 145 insertions(+), 71 deletions(-) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 8ac352442d..244127a4da 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -93,10 +93,11 @@ def order_checkout(self, name: str, problem: str) -> Path: """The checkout thread ``name`` holds for one ordered problem.""" return self.checkout_base.resolve() / f'{name}--{problem}' - def owns_checkout(self, thread: Thread, path: Path) -> bool: - problem = path.name.removeprefix(f'{thread.name}--') - return (re.fullmatch(r'P[1-9][0-9]*', problem) is not None - and path == self.order_checkout(thread.name, problem)) + def owns_checkout(self, name: str, path: Path) -> bool: + problem = path.name.removeprefix(f'{name}--') + return (path == self.checkout_base.resolve() / name + or (re.fullmatch(r'P[1-9][0-9]*', problem) is not None + and path == self.order_checkout(name, problem))) def checkouts_of(self, thread: Thread) -> tuple[Path, ...]: """Read each repository now; directories and names do not register ownership.""" @@ -104,7 +105,7 @@ def checkouts_of(self, thread: Thread) -> tuple[Path, ...]: home = (home if home.exists() else Path(thread.worktree)).expanduser().resolve() ordered = self.order_checkout(thread.name, 'P*') candidates = (home, *(path for path in ordered.parent.glob(ordered.name) - if path.is_dir() and self.owns_checkout(thread, path))) + if path.is_dir() and self.owns_checkout(thread.name, path))) checkouts = {home} listed = set() for repository in candidates: @@ -118,7 +119,7 @@ def checkouts_of(self, thread: Thread) -> tuple[Path, ...]: for field in output.split('\0') if field.startswith('worktree ')) listed.update(entries) checkouts.update(path for path in entries - if path.is_dir() and self.owns_checkout(thread, path)) + if path.is_dir() and self.owns_checkout(thread.name, path)) return (home, *sorted(checkouts - {home})) def deny(self, reason): @@ -193,7 +194,7 @@ def require_path(self, thread: Thread, path): base = self.checkout_base.resolve() if path.is_relative_to(base) and path != base: checkout = base / path.relative_to(base).parts[0] - if not (checkout.name == thread.name or self.owns_checkout(thread, checkout)): + if not self.owns_checkout(thread.name, checkout): self.deny(f'{checkout} is not a checkout of {thread.name}') def render(self, roots): @@ -320,7 +321,7 @@ class CheckoutStorageEffect(AllocateStorageEffect): def check(self, rule, thread): rule.require_path(thread, self.path) - if not rule.owns_checkout(thread, self.path): + if not rule.owns_checkout(thread.name, self.path): destination = rule.order_checkout(thread.name, "P") rule.deny(f'checkout destination must be {destination}') rule.require_allocation(thread, self.path) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 896e0be57f..a3f2d53457 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -203,7 +203,7 @@ async def merge_state(self, todo: Todo, registry: Registration) -> DeliveryMerge if WorkBranches.main in branches: return MainDeliveryMerge() requests = await asyncio.to_thread(PullRequests((todo,)).listed, - todo.repo, "merged", todo.order) + todo.repo, "merged", todo) for request in requests: for merge in await request.delivery_merges(self, check): return merge @@ -296,6 +296,11 @@ class PullRequest(DeclaredFamily, affix="PullRequest"): updated_at: float = 0.0 orders: tuple[str, ...] = () base: str = field(default=WorkBranches.main, kw_only=True) + problems: tuple[str, ...] = field(default=(), kw_only=True) + + def belongs_to(self, todo: Todo) -> bool: + return (todo.id in self.problems + or any(identity in self.orders for identity in todo.held_order_ids)) def merged_commits_since(self, since: float) -> Iterator[str]: return iter(()) @@ -353,7 +358,8 @@ def parsed(self) -> PullRequest: ... class UnmergedListedPullRequest(ListedPullRequest): def parsed(self) -> UnmergedPullRequest: return UnmergedPullRequest(self.number, self.branch, self.head, _github_time(self.updated), - ProblemOrder.pull_request_ids(self.body), base=self.base) + ProblemOrder.pull_request_ids(self.body), base=self.base, + problems=ProblemOrder.pull_request_problems(self.body)) @dataclass(frozen=True, kw_only=True) @@ -364,36 +370,37 @@ class MergedListedPullRequest(ListedPullRequest): def parsed(self) -> MergedPullRequest: return MergedPullRequest(self.number, self.branch, self.head, _github_time(self.updated), ProblemOrder.pull_request_ids(self.body), base=self.base, - merge=self.merge.oid, merged_at=_github_time(self.merged)) + merge=self.merge.oid, merged_at=_github_time(self.merged), + problems=ProblemOrder.pull_request_problems(self.body)) class PullRequests: - """Pull requests queried by immutable order identity, cached within one board read.""" + """Problem and retained order references, cached within one board read.""" def __init__(self, todos: Sequence[Todo] = ()) -> None: self._read: dict[tuple[str, str, str], tuple[PullRequest, ...]] = {} - self._orders: dict[str, dict[str, CheckedProblemOrder]] = {} + self._todos: dict[str, dict[str, Todo]] = {} for todo in todos: if todo.holder: - order = todo.order.checked() - self._orders.setdefault(todo.repo, {})[order.id] = order + self._todos.setdefault(todo.repo, {})[todo.id] = todo - def listed(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, ...]: - checked = order.checked() - key = (repo, state, checked.id) + def listed(self, repo: str, state: str, todo: Todo) -> tuple[PullRequest, ...]: + key = (repo, state, todo.id) if key not in self._read: - self._read[key] = self.read(repo, state, checked) - return tuple(request for request in self._read[key] if checked.id in request.orders) - - def read(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, ...]: - checked = order.checked() - orders = self._orders.setdefault(repo, {}) - orders[checked.id] = checked - aliases = {f"o{index}": candidate for index, candidate in enumerate(orders.values())} + self._read[key] = self.read(repo, state, todo) + return tuple(request for request in self._read[key] if request.belongs_to(todo)) + + def read(self, repo: str, state: str, todo: Todo) -> tuple[PullRequest, ...]: + problems = self._todos.setdefault(repo, {}) + problems[todo.id] = todo + references = ((candidate, reference) for candidate in problems.values() + for reference in (candidate.id, *(order.pull_request_reference + for order in candidate.order.history))) + aliases = {f"o{index}": pair for index, pair in enumerate(references)} selections = [] - for alias, candidate in aliases.items(): + for alias, (_, reference) in aliases.items(): search = json.dumps(f'repo:{repo} is:pr is:{state} in:body ' - f'"{candidate.pull_request_reference}" sort:updated-desc') + f'"{reference}" sort:updated-desc') selections.append( f'{alias}: search(query: {search}, type: ISSUE, first: 100) {{ nodes {{ ' '... on PullRequest { number branch: headRefName head: headRefOid body ' @@ -409,9 +416,12 @@ def read(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, raise WorkBranchError(listed.stderr.strip() or f"GitHub order lookup in {repo} failed.") rows = FieldCodec.decode(dict[str, tuple[ListedPullRequest, ...]], FieldCodec.parse(listed.stdout)) - for alias, candidate in aliases.items(): - self._read[repo, state, candidate.id] = tuple(row.parsed() for row in rows[alias]) - return self._read[repo, state, checked.id] + for candidate in problems.values(): + requests = {row.number: row.parsed() for alias, (problem, _) in aliases.items() + if problem.id == candidate.id for row in rows[alias]} + self._read[repo, state, candidate.id] = tuple(sorted( + requests.values(), key=lambda request: request.updated_at, reverse=True)) + return self._read[repo, state, todo.id] def _clock(at: float) -> str: @@ -523,51 +533,72 @@ class ProgressCheckout(DeclaredFamily, affix="Checkout"): path: Path @abstractmethod - def commits(self, since: float, *excluded: str) -> Iterator[ProgressPoint]: ... + def commits(self, pattern: str, refs: tuple[str, ...], + *excluded: str) -> Iterator[ProgressPoint]: ... + + @abstractmethod + def references(self, storage: StorageRule, owner: str) -> tuple[str, ...]: ... @abstractmethod def readings(self, source: ProgressSource) -> DistanceReadings: ... @abstractmethod - def pushes(self, since: float) -> Iterator[tuple[str, float]]: ... + def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: ... class MissingCheckout(ProgressCheckout): """A retired checkout contributes no local history, not a failed Git read.""" - def commits(self, since: float, *excluded: str) -> Iterator[ProgressPoint]: + def commits(self, pattern: str, refs: tuple[str, ...], + *excluded: str) -> Iterator[ProgressPoint]: return iter(()) + def references(self, storage: StorageRule, owner: str) -> tuple[str, ...]: + return () + def readings(self, source: ProgressSource) -> DistanceReadings: return DistanceReadings() - def pushes(self, since: float) -> Iterator[tuple[str, float]]: + def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: return iter(()) class PresentCheckout(ProgressCheckout): def git(self, *args: str) -> str: - return CheckedCommit.git( - str(self.path), f"--git-dir={(self.path / '.git').absolute()}", *args) + return CheckedCommit.git(str(self.path), *args) - def commits(self, since: float, *excluded: str) -> Iterator[ProgressPoint]: - """Newest branch commit; a Git failure in a present checkout remains an error.""" - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(since)) - log = self.git("log", "-1", f"--since={after}", f"--format={ProgressPoint.format}", - "HEAD", "--not", *excluded) - return (ProgressPoint.listed(line) for line in log.splitlines()) + @cached_property + def origin_refs(self) -> tuple[str, ...]: + return tuple(self.git("for-each-ref", "--format=%(refname)", + "refs/remotes/origin/").splitlines()) + + def references(self, storage: StorageRule, owner: str) -> tuple[str, ...]: + local = tuple(f"refs/heads/{branch}" for branch, paths in WorkBranches(self.path).worktrees().items() + if any(storage.owns_checkout(owner, Path(path).resolve()) for path in paths)) + return (*local, *self.origin_refs) + + def commits(self, pattern: str, refs: tuple[str, ...], + *excluded: str) -> Iterator[ProgressPoint]: + """Newest problem commit across owned branches and retained origin references.""" + if refs: + for line in self.git("log", "-1", "--extended-regexp", "--regexp-ignore-case", + f"--grep={pattern}", f"--format={ProgressPoint.format}", + *refs, "--not", *excluded).splitlines(): + yield ProgressPoint.listed(line) def readings(self, source: ProgressSource) -> DistanceReadings: """Read the bounded distance trend only from explicitly recorded check results.""" - base = self.git("merge-base", "HEAD", "origin/HEAD").strip() - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(source.since)) - commits = [ProgressPoint.listed(line) for line in reversed(self.git( - "log", f"--max-count={DistanceReadings.window}", "--first-parent", f"--since={after}", - f"--format={ProgressPoint.format}", "HEAD", "--not", base).splitlines())] - points = [ProgressPoint(base, source.since), *commits][-DistanceReadings.window:] - order = source.todo.order.checked() - return DistanceReadings(tuple( - reading for point in points for reading in self.recorded_distances(order, base, point))) + readings = [] + for tip in source.commits("origin/HEAD"): + base = self.git("merge-base", tip.sha, "origin/HEAD").strip() + commits = [ProgressPoint.listed(line) for line in reversed(self.git( + "log", f"--max-count={DistanceReadings.window}", "--first-parent", + "--extended-regexp", "--regexp-ignore-case", f"--grep={source.todo.reference_pattern}", + f"--format={ProgressPoint.format}", tip.sha, "--not", base).splitlines())] + points = [ProgressPoint(base, source.since), *commits][-DistanceReadings.window:] + readings.extend(reading for point in points for reading in self.recorded_distances( + source.todo.order.checked(), base, point)) + return DistanceReadings(tuple(readings)) def recorded_distances(self, order: CheckedProblemOrder, base: str, point: ProgressPoint) -> Iterator[DistanceReading]: @@ -577,30 +608,36 @@ def recorded_distances(self, order: CheckedProblemOrder, base: str, yield DistanceReading(point.at, order.check.result.distance(run).plus( order.untouched(touched)), order.check.command) - def pushes(self, since: float) -> Iterator[tuple[str, float]]: - """A push-ref update has its own time, not the commit's author time.""" - branch = self.git("branch", "--show-current").strip() - after = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(since)) - for row in self.git("reflog", "show", "--all", "--date=unix", f"--since={after}", + def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: + """Retained origin commits prove publication; their push reflog supplies its clock.""" + published = {point.sha for point in self.commits(pattern, self.origin_refs)} + for row in self.git("reflog", "show", "--all", "--date=unix", "--grep-reflog=update by push", "--format=%gD%x09%H").splitlines(): selector, sha = row.split("\t") reference, stamp = selector.rsplit("@{", 1) - if reference == f"refs/remotes/origin/{branch}": + if reference in self.origin_refs and sha in published: yield sha, float(stamp.removesuffix("}")) @dataclass(frozen=True) class ProgressSource: - """An order's local checkout capability and independently retained PR references.""" + """Problem progress across owned worktrees, shared Git history and retained PR identities.""" todo: Todo checkout: ProgressCheckout pulls: PullRequests + references: tuple[str, ...] = () @classmethod def of(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> Self: - path = storage.order_checkout(todo.held().owner, todo.id) - checkout = PresentCheckout(path) if path.is_dir() else MissingCheckout(path) - return cls(todo, checkout, pulls) + owner = todo.held().owner + owned = tuple(sorted(path for path in storage.checkout_base.glob(f"{owner}*") + if path.is_dir() and storage.owns_checkout(owner, path.resolve()))) + candidates = (Path(todo.order.checked().check.directory), *owned) + path = next((path for path in candidates if (path / ".git").exists()), + next(iter(owned), candidates[0])) + checkout = (PresentCheckout(path) if path in owned or (path / ".git").exists() + else MissingCheckout(path)) + return cls(todo, checkout, pulls, checkout.references(storage, owner)) @property def since(self) -> float: @@ -608,15 +645,15 @@ def since(self) -> float: def requests(self, state: str) -> list[PullRequest]: """Explicit PR references survive checkout removal or branch reuse.""" - return [pr for pr in self.pulls.listed(self.todo.repo, state, self.todo.order) - if pr.updated_at >= self.since] + return list(self.pulls.listed(self.todo.repo, state, self.todo)) def commits(self, *excluded: str) -> Iterator[ProgressPoint]: - return self.checkout.commits(self.since, *excluded) + return self.checkout.commits(self.todo.reference_pattern, self.references, *excluded) @cached_property def pushes(self) -> tuple[PushedProgress, ...]: - return tuple(PushedProgress(sha[:9], at=at) for sha, at in self.checkout.pushes(self.since)) + return tuple(PushedProgress(sha[:9], at=at) + for sha, at in self.checkout.pushes(self.todo.reference_pattern)) @cached_property def moved_at(self) -> float: @@ -1318,6 +1355,19 @@ class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): finding: ProblemFinding = field(default_factory=UnlocatedProblemFinding, metadata={"wire_omit_default": True}) + @property + def history(self) -> tuple[CheckedProblemOrder, ...]: + return () + + def replaced_by(self, order: PendingProblemOrder) -> PendingProblemOrder: + return order + + @classmethod + def pull_request_problems(cls, body: str) -> tuple[str, ...]: + """Explicit body declarations and problem headings, never titles or incidental mentions.""" + return tuple(re.findall( + r"(?m)^(?:#{1,6} +|Agent-Comms-Problem: +|Problem: +)(P[1-9][0-9]*)(?=[:\s]|$)", body)) + def reissued(self, at: float) -> PendingProblemOrder: """This order sent again at ``at`` to a new owner, its first-call window kept.""" raise ValueError("Only an undelivered order can change owner.") @@ -1401,6 +1451,10 @@ class FoldedProblemOrder(ProblemOrder): def phase(self) -> type[OrderPhase]: return self.original.phase + @property + def history(self) -> tuple[CheckedProblemOrder, ...]: + return self.original.history + def withdrawals(self, todo: Todo) -> tuple[str, ...]: return (f"{todo.id} folded into {self.into}: {self.why}",) @@ -1412,6 +1466,16 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None): def checked(self) -> CheckedProblemOrder: return self + preceding: ProblemOrder = field(default_factory=UnorderedProblemOrder, + metadata={"wire_omit_default": True}) + + @property + def history(self) -> tuple[CheckedProblemOrder, ...]: + return (*self.preceding.history, self) + + def replaced_by(self, order: PendingProblemOrder) -> PendingProblemOrder: + return replace(order, preceding=self.preceding if order.id == self.id else self) + id: str owner: FileLine check: OrderCheck @@ -1493,7 +1557,7 @@ def open_pull_request(self, todo: Todo, checkout: Path, title: str, body: str) - """The order creates its PR with the immutable identity every progress reader uses.""" created = subprocess.run( ("gh", "pr", "create", "-R", todo.repo, "--title", title, "--body-file", "-"), - input=self.pull_request_body(body), cwd=checkout, + input=f"Agent-Comms-Problem: {todo.id}\n{self.pull_request_body(body)}", cwd=checkout, capture_output=True, text=True, timeout=60) if created.returncode: raise ValueError(created.stderr.strip() or "Order pull request creation failed.") @@ -1575,7 +1639,7 @@ def reissued(self, at: float) -> PendingProblemOrder: - self.at.recorded_at("An order")) return replace(self, id=uuid4().hex[:12], at=RecordedEventTime(at), first_call_deadline=RecordedEventTime(at + window), - first_call=AwaitingFirstCall()) + first_call=AwaitingFirstCall(), preceding=self) pending: ClassVar[bool] = True line: ClassVar[str] = "Open" @@ -1955,7 +2019,8 @@ def apply(self, current: Todo) -> Todo: self.order.require_writable() assigned = (current if current.assignment == proposed else current.state.assign(current, proposed, self.traced)) - assigned = replace(assigned, deadline=self.deadline, order=self.order, goal=self.goal, + assigned = replace(assigned, deadline=self.deadline, + order=current.order.replaced_by(self.order), goal=self.goal, revision=current.revision + 1) return self.actor.assigned(current, assigned) @@ -2364,6 +2429,14 @@ class Todo(TypedTable): corrections: tuple[WordingCorrection, ...] = () messages: tuple[OrderMessage, ...] = () + @property + def held_order_ids(self) -> tuple[str, ...]: + return tuple(order.id for order in self.order.history) + + @property + def reference_pattern(self) -> str: + return rf"(^|[^A-Za-z0-9_]){re.escape(self.id)}([^A-Za-z0-9_]|$)" + @property def pending(self): return self.order.pending diff --git a/src/agent_comms/work_branches.py b/src/agent_comms/work_branches.py index 7c997d1459..96be29c54e 100644 --- a/src/agent_comms/work_branches.py +++ b/src/agent_comms/work_branches.py @@ -201,7 +201,7 @@ def _pages(self, query: str, answer: type[ForgeQuery], **variables: str): yield from connection.nodes pages.extend(connection.page_info.following) - def _worktrees(self) -> dict[str, tuple[str, ...]]: + def worktrees(self) -> dict[str, tuple[str, ...]]: """Branch name -> worktree paths that have it checked out.""" checked: dict[str, list[str]] = {} path = None @@ -270,7 +270,7 @@ def read(self, snapshot: RegistrySnapshot) -> tuple[WorkBranch, ...]: ): for linked in issue.linked_branches.nodes: links.setdefault(linked.ref.name, []).append(IssueBranchLink(issue.number)) - worktrees = self._worktrees() + worktrees = self.worktrees() branches = [] for branch, oid in refs.items(): if oid not in commits: From 81e227b461eb7e3e7b82c88e7f9847f3efc86440 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:02:03 -0400 Subject: [PATCH 2/8] P283: derive checkout candidates through StorageRule and keep linked order history --- src/agent_comms/storage_rules.py | 11 ++++++++--- src/agent_comms/todos.py | 5 ++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 244127a4da..3fd158d535 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -99,13 +99,18 @@ def owns_checkout(self, name: str, path: Path) -> bool: or (re.fullmatch(r'P[1-9][0-9]*', problem) is not None and path == self.order_checkout(name, problem))) + def checkout_paths(self, name: str) -> tuple[Path, ...]: + """Candidate locations; Git worktree metadata supplies their branches.""" + ordered = self.order_checkout(name, 'P*') + return tuple(path for path in (self.checkout_base.resolve() / name, + *sorted(ordered.parent.glob(ordered.name))) + if path.is_dir() and self.owns_checkout(name, path)) + def checkouts_of(self, thread: Thread) -> tuple[Path, ...]: """Read each repository now; directories and names do not register ownership.""" home = self.checkout_base / thread.name home = (home if home.exists() else Path(thread.worktree)).expanduser().resolve() - ordered = self.order_checkout(thread.name, 'P*') - candidates = (home, *(path for path in ordered.parent.glob(ordered.name) - if path.is_dir() and self.owns_checkout(thread.name, path))) + candidates = (home, *self.checkout_paths(thread.name)) checkouts = {home} listed = set() for repository in candidates: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index a3f2d53457..86ccc24b4f 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -630,8 +630,7 @@ class ProgressSource: @classmethod def of(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> Self: owner = todo.held().owner - owned = tuple(sorted(path for path in storage.checkout_base.glob(f"{owner}*") - if path.is_dir() and storage.owns_checkout(owner, path.resolve()))) + owned = storage.checkout_paths(owner) candidates = (Path(todo.order.checked().check.directory), *owned) path = next((path for path in candidates if (path / ".git").exists()), next(iter(owned), candidates[0])) @@ -1437,7 +1436,7 @@ def instructions(self, todo: Todo) -> str: *self.finding.instructions(), "No order")) def linked(self, order: PendingProblemOrder) -> PendingProblemOrder: - return replace(order, finding=self.finding) + return self.replaced_by(replace(order, finding=self.finding)) @dataclass(frozen=True, kw_only=True) From 8682cfbd59a05b2ac25e939f4a0c5a8f47ce883b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:05:08 -0400 Subject: [PATCH 3/8] P283: verify retained progress and identify every overdue problem --- src/agent_comms/commander/order.py | 6 +-- src/agent_comms/todos.py | 7 ++- tests/test_order_progress.py | 70 +++++++++++++++++++++++------- 3 files changed, 62 insertions(+), 21 deletions(-) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 33e6ee33de..724c781219 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -664,9 +664,9 @@ def deadline_lines(self, pending: list[Todo], read: dict[str, OrderProgress], continue answer = self.reply(t.holder, ordered) due = f'due {hhmm(t.deadline)}, order {t.order.id}' - late.append(f'{t.holder} ({due}, {work.line}{answer})') - alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now' - return [alarm] * bool(late) + late.append(f'OVERDUE: {t.id} {t.holder} ({due}, {work.line}{answer}) ' + '-> push or merge unfinished work now') + return late def install_lines(self, held: tuple[Todo, ...], read: dict[str, OrderProgress], now: float) -> list[str]: """One INSTALL line naming every held order that waits on the fleet's install.""" diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 86ccc24b4f..164af30241 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1351,6 +1351,7 @@ class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): pending: ClassVar[bool] = False line: ClassVar[str] = "No order" pull_request_label: ClassVar[str] = "Agent-Comms-Order:" + pull_request_problem_label: ClassVar[str] = "Agent-Comms-Problem:" finding: ProblemFinding = field(default_factory=UnlocatedProblemFinding, metadata={"wire_omit_default": True}) @@ -1365,7 +1366,8 @@ def replaced_by(self, order: PendingProblemOrder) -> PendingProblemOrder: def pull_request_problems(cls, body: str) -> tuple[str, ...]: """Explicit body declarations and problem headings, never titles or incidental mentions.""" return tuple(re.findall( - r"(?m)^(?:#{1,6} +|Agent-Comms-Problem: +|Problem: +)(P[1-9][0-9]*)(?=[:\s]|$)", body)) + rf"(?m)^(?:#{{1,6}} +|{re.escape(cls.pull_request_problem_label)} +|Problem: +)" + r"(P[1-9][0-9]*)(?=[:\s]|$)", body)) def reissued(self, at: float) -> PendingProblemOrder: """This order sent again at ``at`` to a new owner, its first-call window kept.""" @@ -1556,7 +1558,8 @@ def open_pull_request(self, todo: Todo, checkout: Path, title: str, body: str) - """The order creates its PR with the immutable identity every progress reader uses.""" created = subprocess.run( ("gh", "pr", "create", "-R", todo.repo, "--title", title, "--body-file", "-"), - input=f"Agent-Comms-Problem: {todo.id}\n{self.pull_request_body(body)}", cwd=checkout, + input=f"{self.pull_request_problem_label} {todo.id}\n{self.pull_request_body(body)}", + cwd=checkout, capture_output=True, text=True, timeout=60) if created.returncode: raise ValueError(created.stderr.strip() or "Order pull request creation failed.") diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index da1e20fce6..8109867175 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -9,19 +9,21 @@ import pytest from agent_comms.recorded_time import RecordedEventTime +from agent_comms.field_codec import FieldCodec +from agent_comms.goal_actions import OrderGoalAction from agent_comms.commander import order as commander_order from agent_comms.child_process import CountPrinted -from agent_comms.goals import CountCheckResult, GitCommandFailed, OrderCheck -from agent_comms.tools import CommsOrderDoneTool +from agent_comms.goals import CountCheckResult, OrderCheck +from agent_comms.tools import CommsOrderDoneTool, CommsOrderTool from agent_comms.todos import (CommittedProgress, DistanceReadings, MergedProgress, NoCommitProgress, InstallDeadline, InstallDue, OpenPrProgress, OperationDelivery, OrderProgress, PlannedInstallDeadline, - MergedPullRequest, PresentCheckout, ProgressPoint, ProgressSource, PullRequest, PullRequests, - PushedProgress, UnmergedPullRequest, UnreadProgress) + MeasuredProgressSource, MergedPullRequest, ProgressPoint, ProgressSource, PullRequest, PullRequests, + PushedProgress, Todo, UnmergedListedPullRequest, UnmergedPullRequest, UnreadProgress) from agent_comms.messages import InfoKind from agent_comms.storage_rules import StorageRule from agent_comms.work_branches import WorkBranchError -from test_commander_problem_orders import create, working +from test_commander_problem_orders import create, order, working class Listed(PullRequests): @@ -120,8 +122,10 @@ def test_uncounted_check_keeps_other_overdue_orders_visible(commander, commander progress = read((comms.goals.todos.get('P1'), unit)) assert progress.line.endswith("; check prints no integer: printf 'no count here'") lines = deadline_lines(script) - alarm = next(line for line in lines if line.startswith('OVERDUE:')) - assert "order order-P1" in alarm and "order order-P2" in alarm + alarm = next(line for line in lines if line.startswith('OVERDUE: P1 ')) + assert len(lines) == 2 + assert 'order order-P1' in alarm + assert any(line.startswith('OVERDUE: P2 ') and 'order order-P2' in line for line in lines) assert "check prints no integer: printf 'no count here'" in alarm assert 'distance 0' not in alarm and 'projected done' not in alarm missing = NoIntegerPrinted() @@ -182,8 +186,43 @@ def test_a_detached_checkout_still_reads_its_order_pr(checkout): assert read(checkout, Listed(open=(UnmergedPullRequest(7, 'fix/p1', 'a' * 40, updated_at=400.0, orders=('order-P1',)),))) == OpenPrProgress(7, 'a' * 9, at=400.0) -def test_an_open_pr_untouched_since_the_order_belongs_to_an_earlier_one(checkout): - assert read(checkout, Listed(open=(UnmergedPullRequest(5, 'fix/p1', 'e' * 40, updated_at=99.0, orders=('order-P1',)),))) == NoCommitProgress() +def test_reused_checkout_replaced_order_and_retired_checkout_keep_problem_progress( + checkout, commander, tmp_path): + first, named = checkout + repository = tmp_path / 'repository' + named.rename(repository) + reused = named.parent / 'worker--P41' + git(repository, 'worktree', 'add', '-q', '-b', 'fix/reused', str(reused)) + sha = commit(reused) + git(reused, 'push', '-q', 'origin', 'HEAD:fix/reused') + assert not named.exists() + assert read((first, reused)).sha == sha + assert isinstance(read((first, reused)), PushedProgress) + + comms, caller = commander + replacement = replace(order(repository, 'replacement-P1'), at=RecordedEventTime(2000)) + CommsOrderTool(owner='worker', order=OrderGoalAction( + problem='P1', order=replacement, request='request', + deadline=RecordedEventTime(3000), commander='lead', expected_revision=first.revision, + assigner=comms.registry.require('lead').incarnation)).apply(comms, caller) + current = comms.goals.todos.get('P1') + assert current.held_order_ids == ('order-P1', 'replacement-P1') + assert current.order.preceding == first.order + assert FieldCodec.decode(Todo, FieldCodec.encode(current)) == current + + bodies = (first.order.pull_request_body('Original working PR'), 'Problem: P1\n') + for body in bodies: + request = UnmergedListedPullRequest(7, 'unrelated-branch-name', 'a' * 40, + updated='1970-01-01T00:00:50Z', body=body, base='main').parsed() + foreign = replace(request, number=8, orders=('order-P10',), problems=('P10',)) + progress = read((current, reused), Listed(open=(foreign, request))) + assert isinstance(progress, OpenPrProgress) and progress.number == 7 + assert progress.line == 'PR #7 open at aaaaaaaaa' + + git(repository, 'worktree', 'remove', str(reused)) + published = read((current, reused)) + assert isinstance(published, PushedProgress) and published.sha == sha + assert not reused.exists() def test_a_pr_merged_before_the_order_is_not_its_merge(checkout): @@ -210,7 +249,8 @@ def test_progress_reads_only_explicit_check_receipts(checkout, tmp_path): check = OrderCheck(str(unit), f'printf x >> {shlex.quote(str(marker))}; printf 0', CountCheckResult(0)) todo = replace(todo, order=replace(todo.order, check=check)) - source = ProgressSource.of(todo, StorageRule(checkout_base=unit.parent), Listed()) + commit(unit) + source = MeasuredProgressSource.of(todo, StorageRule(checkout_base=unit.parent), Listed()) assert not source.readings().readings assert not marker.exists() check.at(git(unit, 'rev-parse', 'HEAD').strip(), str(unit)) @@ -223,10 +263,8 @@ def test_progress_reads_only_explicit_check_receipts(checkout, tmp_path): def test_an_existing_broken_checkout_is_not_silently_treated_as_missing(checkout, tmp_path): bad = tmp_path / 'not-a-repository' / 'worker--P1' bad.mkdir(parents=True) - source = ProgressSource.of(checkout[0], StorageRule(checkout_base=bad.parent), Listed()) - assert isinstance(source.checkout, PresentCheckout) - with pytest.raises(GitCommandFailed): - source.readings() + with pytest.raises(WorkBranchError, match='not a git repository'): + ProgressSource.of(checkout[0], StorageRule(checkout_base=bad.parent), Listed()) def test_board_and_overdue_name_a_merge_waiting_on_install(commander, commander_script, monkeypatch, tmp_path): @@ -346,11 +384,11 @@ def test_overdue_remains_announced_until_recorded(commander, commander_script): polls, since = [], 800.0 for _ in range(3): polls, since = polls + deadline_lines(script, since), time.time() - assert len(polls) == 3 and all(line.startswith('OVERDUE: worker (due ') for line in polls) + assert len(polls) == 3 and all(line.startswith('OVERDUE: P1 worker (due ') for line in polls) comms.messaging.send('worker', 'lead', 'on it', InfoKind).message replied = deadline_lines(script, since) assert len(replied) == 1 and '; REPLIED ' in replied[0] and 'on it' in replied[0] - assert deadline_lines(script, time.time())[0].startswith("OVERDUE:") + assert deadline_lines(script, time.time())[0].startswith("OVERDUE: P1 worker") def test_install_due_reads_the_schedule_time_alone(commander): From 6272c8adb5eae7b934dd37720678217c2242f94e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:09:51 -0400 Subject: [PATCH 4/8] P283: admit only existing Git refs, not unborn worktree branch names --- src/agent_comms/todos.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 164af30241..6f3739eba6 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -568,14 +568,19 @@ def git(self, *args: str) -> str: return CheckedCommit.git(str(self.path), *args) @cached_property - def origin_refs(self) -> tuple[str, ...]: + def git_refs(self) -> tuple[str, ...]: return tuple(self.git("for-each-ref", "--format=%(refname)", - "refs/remotes/origin/").splitlines()) + "refs/heads/", "refs/remotes/origin/").splitlines()) + + @property + def origin_refs(self) -> tuple[str, ...]: + return tuple(ref for ref in self.git_refs if ref.startswith("refs/remotes/origin/")) def references(self, storage: StorageRule, owner: str) -> tuple[str, ...]: - local = tuple(f"refs/heads/{branch}" for branch, paths in WorkBranches(self.path).worktrees().items() - if any(storage.owns_checkout(owner, Path(path).resolve()) for path in paths)) - return (*local, *self.origin_refs) + owned = {f"refs/heads/{branch}" + for branch, paths in WorkBranches(self.path).worktrees().items() + if any(storage.owns_checkout(owner, Path(path).resolve()) for path in paths)} + return (*(ref for ref in self.git_refs if ref in owned), *self.origin_refs) def commits(self, pattern: str, refs: tuple[str, ...], *excluded: str) -> Iterator[ProgressPoint]: @@ -665,7 +670,7 @@ def readings(self) -> DistanceReadings: class MeasuredProgressSource(ProgressSource): - """The board's distance view runs the Check; the stall watcher only reads activity.""" + """The board reads recorded distances; the stall watcher only reads activity.""" def readings(self) -> DistanceReadings: return self.checkout.readings(self) From 1d25ca3e9e3463ab17ca8732cf8e2c5fe037994b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:12:23 -0400 Subject: [PATCH 5/8] P283: type retained distance readings and classify checkout identifiers at the boundary --- src/agent_comms/storage_rules.py | 2 +- src/agent_comms/todos.py | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 3fd158d535..918ee18be4 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -96,7 +96,7 @@ def order_checkout(self, name: str, problem: str) -> Path: def owns_checkout(self, name: str, path: Path) -> bool: problem = path.name.removeprefix(f'{name}--') return (path == self.checkout_base.resolve() / name - or (re.fullmatch(r'P[1-9][0-9]*', problem) is not None + or (bool(re.fullmatch(r'P[1-9][0-9]*', problem)) and path == self.order_checkout(name, problem))) def checkout_paths(self, name: str) -> tuple[Path, ...]: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 6f3739eba6..5efac2fa85 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -593,12 +593,13 @@ def commits(self, pattern: str, refs: tuple[str, ...], def readings(self, source: ProgressSource) -> DistanceReadings: """Read the bounded distance trend only from explicitly recorded check results.""" - readings = [] + readings: list[DistanceReading] = [] for tip in source.commits("origin/HEAD"): base = self.git("merge-base", tip.sha, "origin/HEAD").strip() commits = [ProgressPoint.listed(line) for line in reversed(self.git( "log", f"--max-count={DistanceReadings.window}", "--first-parent", - "--extended-regexp", "--regexp-ignore-case", f"--grep={source.todo.reference_pattern}", + "--extended-regexp", "--regexp-ignore-case", + f"--grep={source.todo.reference_pattern}", f"--format={ProgressPoint.format}", tip.sha, "--not", base).splitlines())] points = [ProgressPoint(base, source.since), *commits][-DistanceReadings.window:] readings.extend(reading for point in points for reading in self.recorded_distances( From dad2e641caef6eb90a8b8263008c8580d3283d29 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 23:27:19 -0400 Subject: [PATCH 6/8] P283: keep retained declarations with the Todo store and use actual PR source --- src/agent_comms/commander/order.py | 13 +++-- src/agent_comms/storage_rules.py | 2 +- src/agent_comms/todos.py | 49 ++++++++++-------- tests/test_order_progress.py | 24 ++++++++- tools/cutover/carry_todo_order_history.py | 63 +++++++++++++++++++++++ 5 files changed, 123 insertions(+), 28 deletions(-) create mode 100644 tools/cutover/carry_todo_order_history.py diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 724c781219..bc5fb92fb0 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -965,16 +965,21 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) class PullRequestVerb(OrderVerb, declared_name="pr"): - help = "open the problem's PR with its required immutable order identity" + help = "open the problem's PR from this owned source checkout with its immutable order identity" pid: str = option("pid") title: str = option("--title") body_file: str = option("--body-file") def run(self, ctx: Commander) -> None: todo = ctx.comms.goals.todos.get(self.pid) - checkout = ctx.comms.registry.storage_rule.order_checkout(todo.holder, todo.id) - print(todo.order.open_pull_request( - todo, checkout, self.title, Path(self.body_file).read_text())) + source = Repo.of(Path.cwd()) + holder = todo.held().owner + if not ctx.comms.registry.storage_rule.owns_checkout(holder, source.checkout): + raise ValueError(f"Run order pr from a source checkout owned by {holder}.") + if source.github != todo.repo: + raise ValueError(f"This source checkout is not the problem repository {todo.repo}.") + print(todo.order.checked().open_pull_request( + todo, source.checkout, self.title, Path(self.body_file).read_text())) @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 918ee18be4..db2a96c047 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -90,7 +90,7 @@ def initial_roots(self, name): return StorageRoots(self.scratch_base / name) def order_checkout(self, name: str, problem: str) -> Path: - """The checkout thread ``name`` holds for one ordered problem.""" + """Allocation path for a new problem checkout; an existing source may be reused.""" return self.checkout_base.resolve() / f'{name}--{problem}' def owns_checkout(self, name: str, path: Path) -> bool: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 5efac2fa85..f8de6d8c52 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -395,7 +395,7 @@ def read(self, repo: str, state: str, todo: Todo) -> tuple[PullRequest, ...]: problems[todo.id] = todo references = ((candidate, reference) for candidate in problems.values() for reference in (candidate.id, *(order.pull_request_reference - for order in candidate.order.history))) + for order in candidate.order_history))) aliases = {f"o{index}": pair for index, pair in enumerate(references)} selections = [] for alias, (_, reference) in aliases.items(): @@ -1362,12 +1362,9 @@ class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): metadata={"wire_omit_default": True}) @property - def history(self) -> tuple[CheckedProblemOrder, ...]: + def declarations(self) -> tuple[CheckedProblemOrder, ...]: return () - def replaced_by(self, order: PendingProblemOrder) -> PendingProblemOrder: - return order - @classmethod def pull_request_problems(cls, body: str) -> tuple[str, ...]: """Explicit body declarations and problem headings, never titles or incidental mentions.""" @@ -1444,7 +1441,7 @@ def instructions(self, todo: Todo) -> str: *self.finding.instructions(), "No order")) def linked(self, order: PendingProblemOrder) -> PendingProblemOrder: - return self.replaced_by(replace(order, finding=self.finding)) + return replace(order, finding=self.finding) @dataclass(frozen=True, kw_only=True) @@ -1459,8 +1456,8 @@ def phase(self) -> type[OrderPhase]: return self.original.phase @property - def history(self) -> tuple[CheckedProblemOrder, ...]: - return self.original.history + def declarations(self) -> tuple[CheckedProblemOrder, ...]: + return self.original.declarations def withdrawals(self, todo: Todo) -> tuple[str, ...]: return (f"{todo.id} folded into {self.into}: {self.why}",) @@ -1473,15 +1470,9 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None): def checked(self) -> CheckedProblemOrder: return self - preceding: ProblemOrder = field(default_factory=UnorderedProblemOrder, - metadata={"wire_omit_default": True}) - @property - def history(self) -> tuple[CheckedProblemOrder, ...]: - return (*self.preceding.history, self) - - def replaced_by(self, order: PendingProblemOrder) -> PendingProblemOrder: - return replace(order, preceding=self.preceding if order.id == self.id else self) + def declarations(self) -> tuple[CheckedProblemOrder, ...]: + return (self,) id: str owner: FileLine @@ -1647,7 +1638,7 @@ def reissued(self, at: float) -> PendingProblemOrder: - self.at.recorded_at("An order")) return replace(self, id=uuid4().hex[:12], at=RecordedEventTime(at), first_call_deadline=RecordedEventTime(at + window), - first_call=AwaitingFirstCall(), preceding=self) + first_call=AwaitingFirstCall()) pending: ClassVar[bool] = True line: ClassVar[str] = "Open" @@ -2028,7 +2019,8 @@ def apply(self, current: Todo) -> Todo: assigned = (current if current.assignment == proposed else current.state.assign(current, proposed, self.traced)) assigned = replace(assigned, deadline=self.deadline, - order=current.order.replaced_by(self.order), goal=self.goal, + order=self.order, prior_orders=current.orders_before(self.order), + goal=self.goal, revision=current.revision + 1) return self.actor.assigned(current, assigned) @@ -2436,10 +2428,21 @@ class Todo(TypedTable): verdicts: tuple[OrderVerdict, ...] = () corrections: tuple[WordingCorrection, ...] = () messages: tuple[OrderMessage, ...] = () + prior_orders: tuple[CheckedProblemOrder, ...] = () + + @property + def order_history(self) -> tuple[CheckedProblemOrder, ...]: + return (*self.prior_orders, *self.order.declarations) + + def orders_before(self, following: ProblemOrder, + additional: tuple[CheckedProblemOrder, ...] = ()) -> tuple[CheckedProblemOrder, ...]: + current = {order.id for order in following.declarations} + return tuple({order.id: order for order in (*self.order_history, *additional) + if order.id not in current}.values()) @property def held_order_ids(self) -> tuple[str, ...]: - return tuple(order.id for order in self.order.history) + return tuple(order.id for order in self.order_history) @property def reference_pattern(self) -> str: @@ -2693,9 +2696,11 @@ def link(self, commander: ThreadIncarnation, problem: str, unit: str, message: s target.state.require_open(target) if target.holder not in (None, unit): raise ValueError(f'{problem} is already held by {target.holder}') - replace(target, order=target.order.linked(source.order), assignment=source.held(), - goal=source.goal, deadline=source.deadline, messages=source.messages, - revision=target.revision + 1).upsert(db) + linked = target.order.linked(source.order) + replace(target, order=linked, + prior_orders=target.orders_before(linked, source.prior_orders), + assignment=source.held(), goal=source.goal, deadline=source.deadline, + messages=source.messages, revision=target.revision + 1).upsert(db) def _select(self, where: str, parameters: tuple) -> tuple[Todo, ...]: """Rows in creation order; an absent store has none, and a read never creates one.""" diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index 8109867175..545ebaffbe 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -15,6 +15,8 @@ from agent_comms.child_process import CountPrinted from agent_comms.goals import CountCheckResult, OrderCheck from agent_comms.tools import CommsOrderDoneTool, CommsOrderTool +from agent_comms.native_tools import NativeRequestTools +from cutover.carry_todo_order_history import TodoOrderHistory from agent_comms.todos import (CommittedProgress, DistanceReadings, MergedProgress, NoCommitProgress, InstallDeadline, InstallDue, OpenPrProgress, OperationDelivery, OrderProgress, PlannedInstallDeadline, MeasuredProgressSource, MergedPullRequest, ProgressPoint, ProgressSource, PullRequest, PullRequests, @@ -58,6 +60,26 @@ def checkout(commander, tmp_path): return create(commander, 'P1'), unit +def test_prior_declarations_do_not_enter_native_request_schemas(): + import json + catalog = json.dumps(NativeRequestTools.catalog()) + assert 'prior_orders' not in catalog + assert 'preceding' not in catalog + + +def test_prior_order_store_carry_preserves_original_rows(commander): + comms, _ = commander + original = create(commander, 'P1') + with comms.goals.todos.writing() as db: + db.execute(f'ALTER TABLE "{Todo.declared_name}" DROP COLUMN "prior_orders"') + preview = TodoOrderHistory.carry(comms.root, apply=False) + assert preview.changed == 2 + applied = TodoOrderHistory.carry(comms.root, apply=True) + assert applied.changed == 2 + assert comms.goals.todos.get('P1') == original + assert TodoOrderHistory.carry(comms.root, apply=True).changed == 0 + + def commit(unit): git(unit, '-c', 'user.name=t', '-c', 'user.email=t@t', 'commit', '-q', '--allow-empty', '-m', 'P1 work') return git(unit, 'rev-parse', 'HEAD')[:9] @@ -207,7 +229,7 @@ def test_reused_checkout_replaced_order_and_retired_checkout_keep_problem_progre assigner=comms.registry.require('lead').incarnation)).apply(comms, caller) current = comms.goals.todos.get('P1') assert current.held_order_ids == ('order-P1', 'replacement-P1') - assert current.order.preceding == first.order + assert current.prior_orders == (first.order,) assert FieldCodec.decode(Todo, FieldCodec.encode(current)) == current bodies = (first.order.pull_request_body('Original working PR'), 'Problem: P1\n') diff --git a/tools/cutover/carry_todo_order_history.py b/tools/cutover/carry_todo_order_history.py new file mode 100644 index 0000000000..f63edc079b --- /dev/null +++ b/tools/cutover/carry_todo_order_history.py @@ -0,0 +1,63 @@ +"""Give the Todo store its prior declarations without exposing them as order inputs.""" + +from contextlib import closing +from pathlib import Path + +from agent_comms.coordination_database import CoordinationStore +from agent_comms.field_codec import FieldCodec +from agent_comms.todos import Todo + +from .carry import Carried, Carry, TodoRows + + +class TodoOrderHistory(TodoRows): + @classmethod + def precedes(cls, later: type[Carry]) -> bool: + return issubclass(later, TodoRows) and later is not cls + + @classmethod + def carry(cls, root: Path, *, apply: bool) -> Carried: + path = cls.store.path_in(root) + if not path.exists(): + return Carried(0) + archive = Todo._fields_named(("prior_orders",))[0] + with closing(CoordinationStore.writer(path)) as db: + db.execute("BEGIN IMMEDIATE") + try: + cursor = db.execute(f'SELECT * FROM "{Todo.declared_name}"') + description = cursor.description + if not description: + raise ValueError("The todo store did not return declared columns") + columns = tuple(column[0] for column in description) + if archive.wire_name in columns: + cls.store(root).require_schema(cls.store._schema(db)) + return Carried(0) + if set(columns) | {archive.wire_name} != set(Todo.columns()): + raise ValueError("The todo store has an unexpected historical schema") + positions = {name: index for index, name in enumerate(columns)} + saved = tuple( + FieldCodec.construct( + Todo, + { + item.name: item.decode(row[positions[item.wire_name]]) + for item in Todo._fields() + if item.wire_name in positions + }, + ) + for row in cursor + ) + if apply: + Todo.drop(db) + Todo.create(db) + for todo in saved: + todo.upsert(db) + cls.store(root).require_schema(cls.store._schema(db)) + db.commit() + return cls.carried(root, len(saved) + 1, {"rows": len(saved)}, apply=apply) + finally: + if db.in_transaction: + db.rollback() + + +if __name__ == "__main__": + TodoOrderHistory.main() From 20a91e8d90e327c633974c3646c46fad203fba01 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 23:35:43 -0400 Subject: [PATCH 7/8] P283: derive progress from actual refs and recorded check evidence --- src/agent_comms/todos.py | 44 ++++++++++++++++++++++++++++++++-------- 1 file changed, 36 insertions(+), 8 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index f8de6d8c52..96347c3411 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -545,6 +545,13 @@ def readings(self, source: ProgressSource) -> DistanceReadings: ... @abstractmethod def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: ... + def checked_commits(self, order: CheckedProblemOrder, refs: tuple[str, ...], + *excluded: str) -> Iterator[ProgressPoint]: + return iter(()) + + def published(self, pattern: str) -> Iterator[ProgressPoint]: + return iter(()) + class MissingCheckout(ProgressCheckout): """A retired checkout contributes no local history, not a failed Git read.""" @@ -564,6 +571,12 @@ def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: class PresentCheckout(ProgressCheckout): + default_reference: ClassVar[str] = "refs/remotes/origin/HEAD" + + @property + def default_refs(self) -> tuple[str, ...]: + return (self.default_reference,) if self.default_reference in self.git_refs else () + def git(self, *args: str) -> str: return CheckedCommit.git(str(self.path), *args) @@ -591,15 +604,24 @@ def commits(self, pattern: str, refs: tuple[str, ...], *refs, "--not", *excluded).splitlines(): yield ProgressPoint.listed(line) + def checked_commits(self, order: CheckedProblemOrder, refs: tuple[str, ...], + *excluded: str) -> Iterator[ProgressPoint]: + """The actual check receipt links a head even without a problem token in its title.""" + return (point for point in self.commits("", refs, *excluded) + if order.check.recorded(point.sha, str(self.path))) + + def published(self, pattern: str) -> Iterator[ProgressPoint]: + return self.commits(pattern, self.origin_refs, *self.default_refs) + def readings(self, source: ProgressSource) -> DistanceReadings: - """Read the bounded distance trend only from explicitly recorded check results.""" + """Read bounded recorded results; an absent remote default supplies no invented base.""" + if not self.default_refs: + return DistanceReadings() readings: list[DistanceReading] = [] - for tip in source.commits("origin/HEAD"): - base = self.git("merge-base", tip.sha, "origin/HEAD").strip() + for tip in source.commits(*self.default_refs): + base = self.git("merge-base", tip.sha, self.default_reference).strip() commits = [ProgressPoint.listed(line) for line in reversed(self.git( "log", f"--max-count={DistanceReadings.window}", "--first-parent", - "--extended-regexp", "--regexp-ignore-case", - f"--grep={source.todo.reference_pattern}", f"--format={ProgressPoint.format}", tip.sha, "--not", base).splitlines())] points = [ProgressPoint(base, source.since), *commits][-DistanceReadings.window:] readings.extend(reading for point in points for reading in self.recorded_distances( @@ -616,7 +638,7 @@ def recorded_distances(self, order: CheckedProblemOrder, base: str, def pushes(self, pattern: str) -> Iterator[tuple[str, float]]: """Retained origin commits prove publication; their push reflog supplies its clock.""" - published = {point.sha for point in self.commits(pattern, self.origin_refs)} + published = {point.sha for point in self.published(pattern)} for row in self.git("reflog", "show", "--all", "--date=unix", "--grep-reflog=update by push", "--format=%gD%x09%H").splitlines(): selector, sha = row.split("\t") @@ -653,7 +675,13 @@ def requests(self, state: str) -> list[PullRequest]: return list(self.pulls.listed(self.todo.repo, state, self.todo)) def commits(self, *excluded: str) -> Iterator[ProgressPoint]: - return self.checkout.commits(self.todo.reference_pattern, self.references, *excluded) + points = (*self.checkout.checked_commits(self.todo.order.checked(), self.references, *excluded), + *self.checkout.commits(self.todo.reference_pattern, self.references, *excluded)) + return iter(sorted({point.sha: point for point in points}.values(), + key=lambda point: point.at, reverse=True)[:1]) + + def published(self) -> Iterator[ProgressPoint]: + return self.checkout.published(self.todo.reference_pattern) @cached_property def pushes(self) -> tuple[PushedProgress, ...]: @@ -832,7 +860,7 @@ class PushedProgress(CommittedProgress): @classmethod def found(cls, source: ProgressSource) -> Iterator[Self]: - moves = (*source.commits("origin/HEAD"), + moves = (*source.published(), *(ProgressPoint(push.sha, push.at) for push in source.pushes)) return (cls(move.sha[:9], at=move.at, readings=source.readings()) for move in sorted(moves, key=lambda move: move.at, reverse=True)[:1]) From 0b3d9cb6e2c03a9b02240d08d5fc37211c9d04a1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 23:39:23 -0400 Subject: [PATCH 8/8] P283: exercise reused source checkout through the Core PR verb --- src/agent_comms/commander/order.py | 3 ++- src/agent_comms/todos.py | 35 +++++++++++++++++++----------- tests/test_order_progress.py | 17 ++++++++++++--- 3 files changed, 38 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index bc5fb92fb0..1b49d79e2a 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -965,7 +965,8 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) class PullRequestVerb(OrderVerb, declared_name="pr"): - help = "open the problem's PR from this owned source checkout with its immutable order identity" + help = ("open the problem's PR from this owned source checkout " + "with its immutable order identity") pid: str = option("pid") title: str = option("--title") body_file: str = option("--body-file") diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 96347c3411..3e439a48d5 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -7,14 +7,14 @@ from __future__ import annotations +import asyncio import calendar import json import re -import subprocess import sqlite3 +import subprocess import sys import time -import asyncio from abc import abstractmethod from collections.abc import Iterator, Sequence from contextlib import contextmanager @@ -22,27 +22,33 @@ from functools import cache, cached_property from importlib.resources import files from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, Final, Self +from typing import TYPE_CHECKING, ClassVar, Self from uuid import uuid4 from . import skills from .change_report import FileLine +from .channel_targets import CommandTarget from .child_process import BoundedRun, CheckPrinted, ChildResult, ExitedOutcome, NoIntegerPrinted from .command import Command from .declared_family import DeclaredFamily from .errors import ToolRefusedError from .field_codec import ( - FieldCodec, GitCommitText, MalformedRecord, NullMember, RecordMember, SubclassUnion, WireValue, + FieldCodec, + GitCommitText, + MalformedRecord, + NullMember, + RecordMember, + SubclassUnion, + WireValue, ) -from .channel_targets import CommandTarget -from .goals import CheckedCommit, CountCheckResult, OrderCheck, ProblemGoalOrder +from .goals import CheckedCommit, OrderCheck, ProblemGoalOrder +from .plan_phases import OrderPhase, UnplannedPhase from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime +from .standing_rules import UndatedStandingRule +from .storage_rules import StorageRule from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation from .threads import Thread from .typed_table import Column, Index, JsonStorage, TypedStore, TypedTable -from .standing_rules import UndatedStandingRule -from .storage_rules import StorageRule -from .plan_phases import OrderPhase, UnplannedPhase from .work_branches import WorkBranchError, WorkBranches if TYPE_CHECKING: @@ -675,8 +681,10 @@ def requests(self, state: str) -> list[PullRequest]: return list(self.pulls.listed(self.todo.repo, state, self.todo)) def commits(self, *excluded: str) -> Iterator[ProgressPoint]: - points = (*self.checkout.checked_commits(self.todo.order.checked(), self.references, *excluded), - *self.checkout.commits(self.todo.reference_pattern, self.references, *excluded)) + points = ( + *self.checkout.checked_commits(self.todo.order.checked(), self.references, *excluded), + *self.checkout.commits(self.todo.reference_pattern, self.references, *excluded), + ) return iter(sorted({point.sha: point for point in points}.values(), key=lambda point: point.at, reverse=True)[:1]) @@ -2462,8 +2470,9 @@ class Todo(TypedTable): def order_history(self) -> tuple[CheckedProblemOrder, ...]: return (*self.prior_orders, *self.order.declarations) - def orders_before(self, following: ProblemOrder, - additional: tuple[CheckedProblemOrder, ...] = ()) -> tuple[CheckedProblemOrder, ...]: + def orders_before( + self, following: ProblemOrder, additional: tuple[CheckedProblemOrder, ...] = (), + ) -> tuple[CheckedProblemOrder, ...]: current = {order.id for order in following.declarations} return tuple({order.id: order for order in (*self.order_history, *additional) if order.id not in current}.values()) diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index 545ebaffbe..4ed7bc73d0 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -320,11 +320,13 @@ def test_order_due_moves_the_deadline_in_core_and_on_the_record(commander, comma class HeldSource: """A source holding an unpushed commit, a pushed commit, an open PR and a merged PR.""" since = 100.0 + pushes = () def commits(self, *excluded): - commits = {('--remotes',): ProgressPoint('a' * 40, 1000.0), - ('origin/HEAD',): ProgressPoint('e' * 40, 1000.0)} - yield commits[excluded] + yield ProgressPoint('a' * 40, 1000.0) + + def published(self): + yield ProgressPoint('e' * 40, 1000.0) def readings(self): return DistanceReadings() @@ -336,10 +338,14 @@ def requests(self, state): class EmptySource: since = 100.0 + pushes = () def commits(self, *excluded): return iter(()) + def published(self): + return iter(()) + def requests(self, state): return [] @@ -469,6 +475,11 @@ def test_unrecorded_merge_uses_order_identity_after_checkout_is_removed( todo = create(commander, pid, deadline=1) storage = StorageRule(checkout_base=tmp_path / "wt") monkeypatch.setattr(script.comms.registry, "storage_rule", storage) + reused = storage.order_checkout("worker", "P999") + reused.mkdir(parents=True) + git(reused, "init", "-q", "-b", branch) + git(reused, "remote", "add", "origin", f"https://github.com/{todo.repo}.git") + monkeypatch.chdir(reused) body_file = tmp_path / "pr-body.md" body_file.write_text(subject) bodies = []