diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 33e6ee33de..1b49d79e2a 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.""" @@ -965,16 +965,22 @@ 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 8ac352442d..db2a96c047 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -90,21 +90,27 @@ 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, 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 (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, ...]: + """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, path))) + candidates = (home, *self.checkout_paths(thread.name)) checkouts = {home} listed = set() for repository in candidates: @@ -118,7 +124,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 +199,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 +326,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..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: @@ -203,7 +209,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 +302,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 +364,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 +376,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 +422,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 +539,100 @@ 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]]: ... + + 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.""" - 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): + 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), f"--git-dir={(self.path / '.git').absolute()}", *args) + return CheckedCommit.git(str(self.path), *args) + + @cached_property + def git_refs(self) -> tuple[str, ...]: + return tuple(self.git("for-each-ref", "--format=%(refname)", + "refs/heads/", "refs/remotes/origin/").splitlines()) - 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()) + @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, ...]: + 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]: + """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 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.""" - 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))) + """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(*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", + 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 +642,35 @@ 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.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") 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 = 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])) + 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 +678,23 @@ 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) + 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, ...]: - 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: @@ -629,7 +707,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) @@ -790,7 +868,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]) @@ -1315,9 +1393,21 @@ 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}) + @property + def declarations(self) -> tuple[CheckedProblemOrder, ...]: + return () + + @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( + 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.""" raise ValueError("Only an undelivered order can change owner.") @@ -1401,6 +1491,10 @@ class FoldedProblemOrder(ProblemOrder): def phase(self) -> type[OrderPhase]: return self.original.phase + @property + 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}",) @@ -1412,6 +1506,10 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None): def checked(self) -> CheckedProblemOrder: return self + @property + def declarations(self) -> tuple[CheckedProblemOrder, ...]: + return (self,) + id: str owner: FileLine check: OrderCheck @@ -1493,7 +1591,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=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.") @@ -1955,7 +2054,9 @@ 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=self.order, prior_orders=current.orders_before(self.order), + goal=self.goal, revision=current.revision + 1) return self.actor.assigned(current, assigned) @@ -2363,6 +2464,26 @@ 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) + + @property + def reference_pattern(self) -> str: + return rf"(^|[^A-Za-z0-9_]){re.escape(self.id)}([^A-Za-z0-9_]|$)" @property def pending(self): @@ -2612,9 +2733,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/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: diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index da1e20fce6..4ed7bc73d0 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -9,19 +9,23 @@ 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.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, - 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): @@ -56,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] @@ -120,8 +144,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 +208,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.prior_orders == (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 +271,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 +285,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): @@ -260,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() @@ -276,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 [] @@ -346,11 +412,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): @@ -409,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 = [] 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()