diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index 1213e59874..912ce09517 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -94,7 +94,7 @@ def packaged_rules_files(cls) -> tuple[Path, ...]: def repo_directory(self, name: str) -> Path: """A configured repository name takes precedence over a directory spelling.""" return next((repo.checkout for repo in self.repos - if repo.github.rsplit('/', 1)[-1] == name), Path(name)).resolve() + if repo.github.named(name)), Path(name)).resolve() def write(self, home: Path) -> Path: """Record the config and start each state file empty, so every tool reads a real file.""" diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 7943f4c17f..6441f23709 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -265,7 +265,7 @@ def read_check(self, line: str) -> OrderCheck: return replace(check, directory=str(directory)) def held_orders(self) -> tuple[Todo, ...]: - return self.comms.goals.todos.held_for_commander(self.incarnation) + return tuple(todo for todo in self.todos() if todo.holder) def held_by(self, unit: str) -> tuple[Todo, ...]: return self.comms.goals.todos.held_by(self.incarnation, unit) @@ -677,9 +677,9 @@ def deadline_lines(self, pending: list[Todo], read: dict[str, OrderProgress], continue answer = self.reply(t.holder, ordered) due = f'due {t.deadline.clock()}, 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.""" @@ -996,18 +996,23 @@ 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 = Repo.of(Path.cwd()) - if checkout.github.lower() != todo.repo.lower(): - sys.exit(f'Refused: current checkout belongs to {checkout.github}, not {todo.repo}.') - print(todo.order.open_pull_request( - todo, checkout.checkout, self.title, Path(self.body_file).read_text())) + source = Repo.of(Path.cwd()) + holder = todo.held().owner_identity + 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.name}.") + if not source.holds(todo): + sys.exit(f'Refused: current checkout belongs to {source.github.to_wire()}, ' + f'not {todo.repo.to_wire()}.') + 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/commander/standing.py b/src/agent_comms/commander/standing.py index 90f69b84c2..86d6fc32b6 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -14,6 +14,7 @@ from ..goal_actions import OrderGoalAction from ..goals import Goal from ..repo import Repo +from ..repository_name import RepositoryName from ..standing_rules import ( DatedStandingRule, ReasonedStandingRule, @@ -214,7 +215,7 @@ class MergeGrowth(DatedStandingRule): limit: ClassVar[int] = 40 window: ClassVar[str] = "12 hours ago" - repo: str + repo: RepositoryName sha: str subject: str merged: float @@ -279,6 +280,7 @@ def began(self) -> float: def line(self) -> str: (a, d), (ta, td) = self.src, self.tests return ( - f"GROWTH {self.repo} {self.sha[:9]} src +{a}/-{d} (net {a - d:+}) tests +{ta}/-{td}: " + f"GROWTH {self.repo.to_wire()} {self.sha[:9]} src +{a}/-{d} (net {a - d:+}) " + f"tests +{ta}/-{td}: " f"{self.subject[:90]} -> justify or send back the owner with a deletion order" ) diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 37eaf479ca..e072db611d 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -38,6 +38,7 @@ ) from .progress_claims import ProgressClaim from .recorded_time import EventTime +from .repository_name import RepositoryName from .thread_identity import ThreadIncarnation from .threads import Thread from .todos import GoalRef, PendingProblemOrder, SendBackVerdict, Todo, TodoOperation @@ -564,7 +565,7 @@ def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Tod @dataclass(frozen=True, kw_only=True) class CreateProblemGoalAction(ProblemOrderGoalAction): - repo: str + repo: RepositoryName text: str def bound_to(self, actor: ThreadIncarnation) -> CreateProblemGoalAction: diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index 79df05b074..b8200aee76 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -4,8 +4,14 @@ import subprocess from pathlib import Path +from typing import TYPE_CHECKING, Annotated -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, InstanceOf, PlainSerializer, PlainValidator + +from .repository_name import RepositoryName + +if TYPE_CHECKING: + from .todos import Todo class Repo(BaseModel): @@ -14,9 +20,18 @@ class Repo(BaseModel): model_config = ConfigDict(frozen=True) checkout: Path - github: str + github: InstanceOf[ + Annotated[ + RepositoryName, + PlainValidator(RepositoryName.from_wire, json_schema_input_type=str), + PlainSerializer(RepositoryName.to_wire), + ] + ] source: str = "src" + def holds(self, todo: Todo) -> bool: + return self.github == todo.repo + def contains(self, sha: str, revision: str = "HEAD") -> bool: """Prove ancestry in the requested fetched revision without moving the checkout.""" result = subprocess.run( @@ -43,4 +58,4 @@ def of(cls, checkout: Path) -> Repo: """Its GitHub name comes from the checkout's own origin remote.""" url = cls.git(checkout, "remote", "get-url", "origin").strip() github = url.removesuffix(".git").replace(":", "/").split("/")[-2:] - return cls(checkout=checkout.resolve(), github="/".join(github)) + return cls(checkout=checkout.resolve(), github=RepositoryName.from_wire("/".join(github))) diff --git a/src/agent_comms/repository_name.py b/src/agent_comms/repository_name.py new file mode 100644 index 0000000000..327e7da59a --- /dev/null +++ b/src/agent_comms/repository_name.py @@ -0,0 +1,49 @@ +"""A GitHub repository identity, with its existing owner/name wire form.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import ClassVar + +from .field_codec import MalformedRecord, WireValue + + +@dataclass(frozen=True, slots=True, eq=False) +class RepositoryName(WireValue): + owner: str + name: str + _part: ClassVar[re.Pattern[str]] = re.compile(r"(?!\.{1,2}$)[\w.-]+\Z") + + def __post_init__(self) -> None: + if len(self.to_wire()) > 200 or not all(self._part.fullmatch(part) for part in self.key): + raise ValueError("Repo ID must have the exact owner/repo form.") + + @property + def key(self) -> tuple[str, str]: + return self.owner.casefold(), self.name.casefold() + + def __eq__(self, other: object) -> bool: + if not isinstance(other, RepositoryName): + return NotImplemented + return self.key == other.key + + def __hash__(self) -> int: + return hash(self.key) + + def named(self, name: str) -> bool: + return self.name.casefold() == name.casefold() + + def to_wire(self) -> str: + return "/".join(self.key) + + @classmethod + def from_wire(cls, data: object) -> RepositoryName: + if not isinstance(data, str): + raise MalformedRecord("Repo ID must be an owner/repo string.") + owner, _, name = data.partition("/") + return cls(owner, name) + + @classmethod + def schema(cls) -> dict[str, object]: + return {"type": "string", "description": "GitHub repository owner/name"} diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 66115fbd6d..65ed09af43 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -26,6 +26,7 @@ if TYPE_CHECKING: from .comms import Comms + from .thread_identity import ThreadIncarnation from .threads import Thread StoragePath = Annotated[Path, PathText] @@ -93,21 +94,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, owner: ThreadIncarnation, path: Path) -> bool: + problem = path.name.removeprefix(f'{owner.name}--') + return (path == self.checkout_base.resolve() / owner.name + or (bool(re.fullmatch(r'P[1-9][0-9]*', problem)) + and path == self.order_checkout(owner.name, problem))) + + def checkout_paths(self, owner: ThreadIncarnation) -> tuple[Path, ...]: + """Candidate locations; Git worktree metadata supplies their branches.""" + ordered = self.order_checkout(owner.name, 'P*') + return tuple(path for path in (self.checkout_base.resolve() / owner.name, + *sorted(ordered.parent.glob(ordered.name))) + if path.is_dir() and self.owns_checkout(owner, path)) def checkouts_of(self, thread: Thread) -> tuple[Path, ...]: """Read each repository now; directories and names do not register ownership.""" home = self.checkout_base.expanduser().resolve() / thread.name - ordered = self.order_checkout(thread.name, 'P*') candidates = (home, Path(thread.worktree).expanduser().resolve(), - *(path for path in ordered.parent.glob(ordered.name) - if path.is_dir() and self.owns_checkout(thread, path))) + *self.checkout_paths(thread.incarnation)) checkouts = {home} listed = set() for repository in candidates: @@ -121,7 +128,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.incarnation, path)) return (home, *sorted(checkouts - {home})) def deny(self, reason): @@ -199,7 +206,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.incarnation, checkout): self.deny(f'{checkout} is not a checkout of {thread.name}') def render(self, roots): @@ -359,7 +366,7 @@ class CheckoutStorageEffect(AllocateStorageEffect): def check(self, projection: StorageProjection): rule, thread = projection.rule, projection.thread rule.require_path(thread, self.path) - if not rule.owns_checkout(thread, self.path): + if not rule.owns_checkout(thread.incarnation, self.path): destination = rule.order_checkout(thread.name, "P") rule.deny(f'checkout destination must be {destination}') super().check(projection) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 4a7786ffa3..6d9105eeed 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,33 +22,40 @@ 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 .bus_publication import AgentOrigin +from .channel_targets import CommandTarget from .child_process import ( BoundedRun, CheckPrinted, ChildResult, ExitedOutcome, NoIntegerPrinted, OrderCheckFailedError, ) -from .bus_publication import AgentOrigin from .external_transcript import ExternalTranscript from .messages import InfoKind, Message 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 .order_story import ProblemStory, StoryEvent, StoryLine, StorySource +from .plan_phases import OrderPhase, UnplannedPhase from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime -from .thread_identity import RecordedThreadIncarnation, StallWatcherIncarnation, ThreadIncarnation -from .threads import Thread -from .typed_table import Column, Index, JsonStorage, TypedStore, TypedTable +from .repository_name import RepositoryName from .standing_rules import UndatedStandingRule from .storage_rules import StorageRule -from .plan_phases import OrderPhase, UnplannedPhase +from .thread_identity import RecordedThreadIncarnation, StallWatcherIncarnation, ThreadIncarnation +from .threads import Thread +from .typed_table import Column, Index, JsonStorage, TextStorage, TypedStore, TypedTable from .work_branches import WorkBranchError, WorkBranches if TYPE_CHECKING: @@ -216,7 +223,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 @@ -309,8 +316,13 @@ 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 merged_commits_since(self, since: float) -> Iterator[str]: + 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(self) -> Iterator[str]: return iter(()) async def delivery_merges(self, delivery: CommitDelivery, @@ -327,9 +339,8 @@ class MergedPullRequest(PullRequest): merge: str merged_at: float - def merged_commits_since(self, since: float) -> Iterator[str]: - if self.merged_at >= since: - yield self.merge + def merged_commits(self) -> Iterator[str]: + yield self.merge async def delivery_merges(self, delivery: CommitDelivery, check: OrderCheck) -> tuple[DeliveryMerge, ...]: @@ -366,7 +377,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) @@ -377,36 +389,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._read: dict[tuple[RepositoryName, str, str], tuple[PullRequest, ...]] = {} + self._todos: dict[RepositoryName, 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: RepositoryName, 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: RepositoryName, 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(): - search = json.dumps(f'repo:{repo} is:pr is:{state} in:body ' - f'"{candidate.pull_request_reference}" sort:updated-desc') + for alias, (_, reference) in aliases.items(): + search = json.dumps(f'repo:{repo.to_wire()} is:pr is:{state} in:body ' + 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 ' @@ -419,12 +432,16 @@ def read(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, f'. + {{kind: "{MergedListedPullRequest.declared_name}"}} end))'), capture_output=True, text=True, timeout=60) if listed.returncode: - raise WorkBranchError(listed.stderr.strip() or f"GitHub order lookup in {repo} failed.") + raise WorkBranchError( + listed.stderr.strip() or f"GitHub order lookup in {repo.to_wire()} 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] @dataclass(frozen=True) @@ -532,51 +549,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: ThreadIncarnation) -> 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: ThreadIncarnation) -> 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: ThreadIncarnation) -> 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]: + """A recorded check 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]: @@ -586,30 +652,34 @@ 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_identity + 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 / ".git").exists() else MissingCheckout(path) + return cls(todo, checkout, pulls, checkout.references(storage, owner)) @property def since(self) -> float: @@ -617,15 +687,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: @@ -638,7 +716,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) @@ -808,7 +886,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]) @@ -846,7 +924,7 @@ class MergedProgress(FoundProgress): @classmethod def found(cls, source: ProgressSource) -> Iterator[Self]: return (cls(sha) for pr in source.requests("merged") - for sha in pr.merged_commits_since(source.since)) + for sha in pr.merged_commits()) @property def state(self) -> str: @@ -1350,6 +1428,7 @@ class ProblemOrder(DeclaredFamily, StorySource, 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}) @@ -1358,6 +1437,21 @@ def observe_objective(self, failure: type[ObjectiveStrike], goals: Goals, todo: """Completed or retired orders do not acquire new pending-work failures.""" return () + @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 closed_by_merge(self, todo: Todo, repos: Sequence[Repo], storage: StorageRule, + pulls: PullRequests) -> tuple[Todo, ...]: + return () + 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.") @@ -1444,6 +1538,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}",) @@ -1466,6 +1564,10 @@ def require_wording(self) -> None: def story(self) -> tuple[StoryLine, ...]: return (StoryLine(self.at, f"Ordered {self.id}: {'; '.join(self.edits)}"),) + @property + def declarations(self) -> tuple[CheckedProblemOrder, ...]: + return (self,) + id: str owner: FileLine check: OrderCheck @@ -1537,6 +1639,14 @@ def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind: def merged(self, sha: str) -> MergedProblemOrder: return self.copy_as(MergedProblemOrder, sha=sha) + def closed_by_merge(self, todo: Todo, repos: Sequence[Repo], storage: StorageRule, + pulls: PullRequests) -> tuple[Todo, ...]: + if not repos: + return () + source = ProgressSource.of(todo, storage, pulls) + return tuple(DoneTodoState.closed(todo, merged) for repo in repos + for merged in self.observed_merges(repo, source)) + def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[ProblemOrder, ...]: return tuple(self.merged(merged.sha) for merged in MergedProgress.found(source) if repo.contains(merged.sha)) @@ -1556,8 +1666,9 @@ def pull_request_body(self, body: str) -> str: def open_pull_request(self, todo: Todo, checkout: Path, title: str, body: str) -> 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, + ("gh", "pr", "create", "-R", todo.repo.to_wire(), "--title", title, "--body-file", "-"), + 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.") @@ -2032,7 +2143,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) @@ -2267,7 +2380,7 @@ def reviewed(self, goals: Goals, problem: str, reviewer: ThreadIncarnation) -> T current = goals.todos.get(problem) recorded = self.prepare(goals, current).record(goals, problem) posted = subprocess.run( - ("gh", "api", "--method", "POST", f"repos/{recorded.repo}/statuses/{self.head}", + ("gh", "api", "--method", "POST", f"repos/{recorded.repo.to_wire()}/statuses/{self.head}", "-f", "state=success", "-f", f"context={self.context}", "-f", f"description=Accepted by {reviewer.name} on {problem}"), capture_output=True, text=True, timeout=60, @@ -2545,7 +2658,7 @@ class Todo(TypedTable): indexes = (Index(("creator", "creator_created", "state")), Index(("state",))) id: str = field(metadata={"sql": Column(primary_key=True)}) - repo: str + repo: RepositoryName = field(metadata={"sql": Column(storage=TextStorage)}) text: str creator: str creator_created: float @@ -2561,11 +2674,31 @@ 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_]|$)" unrequested_goal: ClassVar[GoalReference] = NoGoalReference() @classmethod - def ordered(cls, todo_id: str, repo: str, text: str, creator: ThreadIncarnation, + def ordered(cls, todo_id: str, repo: RepositoryName, text: str, creator: ThreadIncarnation, assignment: Assignment, deadline: EventTime, order: PendingProblemOrder | MergedProblemOrder, goal: GoalReference = unrequested_goal) -> Todo: @@ -2573,7 +2706,7 @@ def ordered(cls, todo_id: str, repo: str, text: str, creator: ThreadIncarnation, _text(text, "Todo text", 2000) deadline.require_deadline() goal.require_reference() - return cls(todo_id, _repo(repo), text, creator.name, creator.wire_birth, OpenTodoState, + return cls(todo_id, repo, text, creator.name, creator.wire_birth, OpenTodoState, 1, goal, assignment, order, deadline) @property @@ -2746,17 +2879,6 @@ def _participant(thread: Thread) -> ThreadIncarnation: return thread.incarnation -def _repo(value: str) -> str: - _text(value, "Repo ID", 200) - pieces = value.split("/") - if len(pieces) != 2 or any( - piece in {"", ".", ".."} or not all(char.isalnum() or char in "-_." for char in piece) - for piece in pieces - ): - raise TodoError("Repo ID must have the exact owner/repo form.") - return value.lower() - - @dataclass(frozen=True, slots=True) class TodoStore(TypedStore): """One SQLite row owns both todo state and its exclusive assignment. @@ -2835,9 +2957,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.""" @@ -2872,8 +2996,8 @@ def get(self, todo_id: str) -> Todo: _text(todo_id, "Todo ID", 128) return self._one(self._select("id=?", (todo_id,)), todo_id) - def list(self, repo: str) -> tuple[Todo, ...]: - return self._select("repo=?", (_repo(repo),)) + def list(self, repo: RepositoryName) -> tuple[Todo, ...]: + return self._select("repo=?", (repo.to_wire(),)) def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: return self._select( @@ -2910,14 +3034,12 @@ def reassignable(self, storage: StorageRule) -> tuple[Todo, ...]: def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], storage: StorageRule) -> None: """Read before locking; close only orders whose observed revision still holds.""" - checked = tuple((todo, order) for todo in self.held_for_commander(commander) - if isinstance(order := todo.order, CheckedProblemOrder)) - pulls = PullRequests(tuple(todo for todo, _ in checked)) - completed = tuple(DoneTodoState.closed(todo, merged) - for todo, order in checked - for repo in repos if repo.github.lower() == todo.repo - for merged in order.observed_merges( - repo, ProgressSource.of(todo, storage, pulls))) + held = self.held_for_commander(commander) + pulls = PullRequests(held) + completed = tuple(closed for todo in held + for closed in todo.order.closed_by_merge( + todo, tuple(repo for repo in repos if repo.holds(todo)), + storage, pulls)) if not completed: return with self.writing() as db: @@ -2982,7 +3104,7 @@ def _open(self, where: str, parameters: tuple, def create( self, todo_id: str, - repo: str, + repo: RepositoryName, text: str, *, creator: Thread, diff --git a/src/agent_comms/work_branches.py b/src/agent_comms/work_branches.py index 7c997d1459..09c49faba9 100644 --- a/src/agent_comms/work_branches.py +++ b/src/agent_comms/work_branches.py @@ -186,8 +186,8 @@ def _run(self, *command: str, input: str = "") -> str: @property def repository(self) -> tuple[str, str]: - owner, name = Repo.of(self.checkout).github.split("/") - return owner, name + repository = Repo.of(self.checkout).github + return repository.owner, repository.name def _pages(self, query: str, answer: type[ForgeQuery], **variables: str): pages: list[tuple[str, ...]] = [()] @@ -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/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index cdcac546fb..70c9b71d59 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -22,7 +22,8 @@ from .native_settings import NativeThreadDefaults from .phase_tasks import PhaseTask from .recorded_time import RecordedEventTime -from .storage_rules import CheckoutStorageEffect +from .repository_name import RepositoryName +from .storage_rules import CheckoutStorageEffect, StorageProjection from .thread_status import StoppedThreadStatus if TYPE_CHECKING: @@ -63,7 +64,7 @@ def start(self, comms: Comms) -> Thread: ) self.action.request_ref(GoalActionContext(comms.goals, candidate)) comms.registry.require_declaration_available(candidate, StoppedThreadStatus()) - CheckoutStorageEffect(checkout).check(storage, candidate) + CheckoutStorageEffect(checkout).check(StorageProjection(storage, candidate)) subprocess.run( ("git", "-C", self.action.order.check.directory, "fetch", "origin", "main"), check=True, @@ -119,7 +120,7 @@ class CheckedPhaseTask(PhaseTask): """An explicitly authored task; its order owns source, check, edits and run.""" problem: str - repo: str + repo: RepositoryName text: str order: PendingProblemOrder deadline_seconds: float diff --git a/tests/test_commander_edit_guard.py b/tests/test_commander_edit_guard.py index 722c3b2bdc..3dc7f2bf33 100644 --- a/tests/test_commander_edit_guard.py +++ b/tests/test_commander_edit_guard.py @@ -1,6 +1,7 @@ """Real hook entrypoints and /proc session classification; no classifier substitution.""" from __future__ import annotations +from agent_comms.repository_name import RepositoryName import json import os @@ -22,7 +23,7 @@ def guard_entry(commander, commander_script, tmp_path): context, repo = commander_script context = context.model_copy( update={ - "repos": (Repo(checkout=repo, github="OpenHCSDev/agent-comms"),), + "repos": (Repo(checkout=repo, github=RepositoryName.from_wire("OpenHCSDev/agent-comms")),), } ) context.write(tmp_path) diff --git a/tests/test_commander_problem_orders.py b/tests/test_commander_problem_orders.py index 817820add4..39b0a2acf0 100644 --- a/tests/test_commander_problem_orders.py +++ b/tests/test_commander_problem_orders.py @@ -31,6 +31,7 @@ from agent_comms.goals import CountCheckResult, OrderCheck, ProblemGoalOrder from agent_comms.progress_claims import WorkingClaim from agent_comms.recorded_time import RecordedEventTime, UnrecordedEventTime +from agent_comms.repository_name import RepositoryName from agent_comms.threads import Thread from agent_comms.tool_caller import ThreadCaller from agent_comms.tools import (CommsCorrectWordingTool, CommsSendTool, CommsFireTool, CommsOrderCheckTool, CommsOrderDoneTool, CommsOrderTool, @@ -56,7 +57,7 @@ def create(commander, problem, command='printf 1', expected=1, deadline=1000, ow CommsTodoCreateTool(owner=owner, order=CreateProblemGoalAction( problem=problem, order=order(comms.root, 'order-' + problem, command, expected), request='request', deadline=RecordedEventTime(deadline), commander='lead', - repo='OpenHCSDev/agent-comms', text='Fix ' + problem, + repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Fix ' + problem, )).apply(comms, caller) return comms.goals.todos.get(problem) @@ -76,9 +77,9 @@ def test_new_native_order_refuses_soft_prose_before_reservation(commander, text) CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction( problem='Psoft', order=checked, request='request', deadline=RecordedEventTime(1000), commander='lead', - repo='OpenHCSDev/agent-comms', text='Repair the store', + repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Repair the store', )).apply(comms, caller) - assert comms.goals.todos.list('OpenHCSDev/agent-comms') == () + assert comms.goals.todos.list(RepositoryName.from_wire('OpenHCSDev/agent-comms')) == () assert comms.registry.require('worker').goal == original @@ -107,12 +108,12 @@ def test_direct_creation_checks_defect_prose_before_writing(commander): lead = comms.registry.require('lead') with pytest.raises(ValueError, match='Soft wording'): comms.goals.todos.create( - 'Psoft', 'OpenHCSDev/agent-comms', 'Please repair the store', + 'Psoft', RepositoryName.from_wire('OpenHCSDev/agent-comms'), 'Please repair the store', creator=lead, owner=comms.registry.require('worker'), generation='ongoing', deadline=RecordedEventTime(1000), order=order(comms.root, 'soft-order'), goal=GoalRef(lead.name, lead.created_at, 'request'), ) - assert comms.goals.todos.list('OpenHCSDev/agent-comms') == () + assert comms.goals.todos.list(RepositoryName.from_wire('OpenHCSDev/agent-comms')) == () def working(comms): @@ -740,7 +741,7 @@ def test_a_deletion_order_must_check_the_imports_of_every_repo_that_imports_the_ (toad / 'src').mkdir(parents=True) (toad / 'src' / 'app.py').write_text('from agent_comms.acp_ingress import Ingress\n') script = script.model_copy(update={'repos': ( - Repo(checkout=toad, github='OpenHCSDev/toad'),)}) + Repo(checkout=toad, github=RepositoryName.from_wire('OpenHCSDev/toad')),)}) (repo / 'acp_ingress.py').write_text('class Ingress: ...\n') edit = 'Remove the obsolete ingress implementation.' deletes = ModuleDeletion(('src/agent_comms/acp_ingress.py',)) @@ -1146,7 +1147,8 @@ def stray(commander, problem, generation): not the unit's open Goal.""" comms, _ = commander lead = comms.registry.require('lead') - return comms.goals.todos.create(problem, 'OpenHCSDev/agent-comms', 'Fix ' + problem, creator=lead, + return comms.goals.todos.create( + problem, RepositoryName.from_wire('OpenHCSDev/agent-comms'), 'Fix ' + problem, creator=lead, owner=comms.registry.require('worker'), generation=generation, deadline=RecordedEventTime(1000), order=order(comms.root, 'order-' + problem), goal=GoalRef('lead', lead.created_at, 'request')) @@ -1282,7 +1284,7 @@ def test_dropped_problem_links_through_the_unordered_lifecycle(commander): problem=pid, order=replace(order(comms.root, 'order-' + pid), at=RecordedEventTime(at), first_call_deadline=RecordedEventTime(at + 120), finding=finding), request='request', deadline=RecordedEventTime(1000), commander='lead', - repo='OpenHCSDev/agent-comms', text='Fix ' + pid)).apply(comms, caller) + repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Fix ' + pid)).apply(comms, caller) old_message = order_message(commander, 'P1', 'The earlier counter order') new_message = order_message(commander, 'P2', 'The newer counter order') lead = comms.registry.require('lead').incarnation @@ -1539,7 +1541,7 @@ def test_close_quotes_the_works_line(commander): run = ApplicationOrderRun('The board shows the problem closed after the commander reopens the session') CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction( problem='P1', order=replace(order(comms.root, 'order-P1'), run=run), request='request', - deadline=RecordedEventTime(1000), commander='lead', repo='OpenHCSDev/agent-comms', text='Fix P1', + deadline=RecordedEventTime(1000), commander='lead', repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Fix P1', )).apply(comms, caller) CommsOrderDoneTool(problem='P1', order_id=comms.goals.todos.get('P1').order.id, delivery=OperationDelivery(), claims=working(comms)).apply(comms, caller) artifact = comms.root / 'installed-evidence' @@ -1570,7 +1572,7 @@ def test_delivery_that_renames_the_decision_is_refused(commander, tmp_path): CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction( problem='P520', order=replace(order(comms.root, 'order-P520'), deletes=deletes), request='request', deadline=RecordedEventTime(1000), commander='lead', - repo='OpenHCSDev/agent-comms', text='Delete the is_set decision', + repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Delete the is_set decision', )).apply(comms, caller) pending = comms.goals.todos.get('P520') assert 'Deletes: is_set absence decision at' in pending.order.instructions(pending) diff --git a/tests/test_commander_run_rules.py b/tests/test_commander_run_rules.py index e75d7aaa2f..d0d01236be 100644 --- a/tests/test_commander_run_rules.py +++ b/tests/test_commander_run_rules.py @@ -6,6 +6,7 @@ from agent_comms.change_report import FileLine from agent_comms.field_codec import FieldCodec from agent_comms.recorded_time import RecordedEventTime +from agent_comms.repository_name import RepositoryName from agent_comms.todos import ApplicationOrderRun, AssignTodo, GoalRef, MechanicalOrderRun, Todo, TodoStore from test_todos import order, thread @@ -15,7 +16,8 @@ def test_declared_run_round_trips_and_short_runs_are_refused(tmp_path): checked = replace(order(tmp_path), run=ApplicationOrderRun(planned)) store = TodoStore(tmp_path) lead = thread('lead', tmp_path, 1) - todo = store.create('P1', 'OpenHCSDev/agent-comms', 'Repair the thread list', + todo = store.create('P1', RepositoryName.from_wire('OpenHCSDev/agent-comms'), + 'Repair the thread list', creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), deadline=RecordedEventTime(1000), order=checked) assert FieldCodec.decode(Todo, FieldCodec.encode(todo)) == todo @@ -27,7 +29,8 @@ def test_declared_run_round_trips_and_short_runs_are_refused(tmp_path): def test_clean_transfers_keep_history_and_third_owner_needs_a_real_line(tmp_path): store = TodoStore(tmp_path) lead = thread('lead', tmp_path, 1) - first = store.create('P1', 'OpenHCSDev/agent-comms', 'Repair the thread list', + first = store.create('P1', RepositoryName.from_wire('OpenHCSDev/agent-comms'), + 'Repair the thread list', creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), deadline=RecordedEventTime(1000), order=order(tmp_path)) second = store.assign('P1', expected_revision=first.revision, diff --git a/tests/test_installed_merge.py b/tests/test_installed_merge.py index 7618ebd849..cd742e4afb 100644 --- a/tests/test_installed_merge.py +++ b/tests/test_installed_merge.py @@ -1,4 +1,5 @@ """Merge observation closes the real stored order, without a completion request.""" +from agent_comms.repository_name import RepositoryName from dataclasses import fields, replace import json import subprocess @@ -155,4 +156,4 @@ def test_only_checked_orders_are_selected_for_merge_reconciliation(commander, ch def test_failed_git_is_not_a_negative_ancestry_result(tmp_path): with pytest.raises(subprocess.CalledProcessError): - Repo(checkout=tmp_path / 'missing', github='test/repo').contains('a' * 40) + Repo(checkout=tmp_path / 'missing', github=RepositoryName.from_wire('test/repo')).contains('a' * 40) diff --git a/tests/test_order_declared_deletions.py b/tests/test_order_declared_deletions.py index a5c17e6669..70e0e359ec 100644 --- a/tests/test_order_declared_deletions.py +++ b/tests/test_order_declared_deletions.py @@ -1,4 +1,5 @@ """Declared module deletions reach the real commander CLI and saved order codec.""" +from agent_comms.repository_name import RepositoryName import json import os import subprocess @@ -49,7 +50,7 @@ def test_order_send_cli_refuses_declared_module_imported_by_toad( toad = tmp_path / 'toad' (toad / 'src').mkdir(parents=True) (toad / 'src' / 'app.py').write_text('from agent_comms.acp_ingress import Ingress\n') - script = script.model_copy(update={'repos': (Repo(checkout=toad, github='OpenHCSDev/toad'),)}) + script = script.model_copy(update={'repos': (Repo(checkout=toad, github=RepositoryName.from_wire('OpenHCSDev/toad')),)}) script.write(tmp_path) body = Path(order_body(repo, edit='Remove the obsolete ingress implementation.', deletes=ModuleDeletion(('src/agent_comms/acp_ingress.py',)))) diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index da1e20fce6..deefcfb470 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] @@ -77,7 +101,7 @@ def test_unfinished_check_keeps_other_overdue_orders_visible( monkeypatch.setattr(OrderCheck, "timeout", 0.05) todo = replace(todo, order=replace(todo.order, check=OrderCheck(str(unit), "sleep 0.2; echo 0", CountCheckResult(0)))) - monkeypatch.setattr(script.COMMS.registry, "storage_rule", StorageRule(checkout_base=unit.parent)) + monkeypatch.setattr(script.comms.registry, "storage_rule", StorageRule(checkout_base=unit.parent)) # The child really exceeds its timeout; partial stdout must never become a count. outcome = TimedOutOutcome(SignaledOutcome(9)) assert isinstance(outcome.printed_count(b"0"), NoCountYetPrinted) @@ -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.to_wire()}.git") + monkeypatch.chdir(reused) body_file = tmp_path / "pr-body.md" body_file.write_text(subject) bodies = [] diff --git a/tests/test_order_story.py b/tests/test_order_story.py index b4dda0dc42..71ea67ec5b 100644 --- a/tests/test_order_story.py +++ b/tests/test_order_story.py @@ -1,4 +1,5 @@ """Native stories use one copied Todo record, including a resent send-back.""" +from agent_comms.repository_name import RepositoryName from dataclasses import replace @@ -31,7 +32,7 @@ def test_order_story_copied_bus_strike_and_send_back_once(comms, tmp_path): caller.invoke(CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction( problem="P1", order=replace(order(tmp_path), edits=("Repair the existing record owner",)), request="request", deadline=RecordedEventTime(1000), - commander="lead", repo="org/repo", text="Repair the worker")), comms) + commander="lead", repo=RepositoryName.from_wire("org/repo"), text="Repair the worker")), comms) todo = comms.goals.todos.get("P1") actor = comms.registry.require("lead").incarnation reason = "The worker changed the wrong owner and left the decision duplicated." diff --git a/tests/test_repository_name.py b/tests/test_repository_name.py new file mode 100644 index 0000000000..7485b4805d --- /dev/null +++ b/tests/test_repository_name.py @@ -0,0 +1,53 @@ +"""Repository identities cross the wire and SQL once, without case-sensitive aliases.""" +from dataclasses import replace + +import pytest + +from agent_comms.commander.config import Repo +from agent_comms.field_codec import FieldCodec +from agent_comms.repository_name import RepositoryName +from agent_comms.todos import PullRequests, UnorderedProblemOrder +from test_commander_problem_orders import create + + +def test_repository_identity_hash_and_existing_wire_form(): + first = RepositoryName("OpenHCSDev", "Agent-Comms") + second = RepositoryName.from_wire("OPENHCSDEV/agent-comms") + assert first == second + assert len({first, second}) == 1 + assert FieldCodec.encode(first) == "openhcsdev/agent-comms" + assert FieldCodec.decode(RepositoryName, FieldCodec.encode(first)) == first + + +@pytest.mark.parametrize("owner", ["ß" * 100, "İ"]) +def test_repository_rejects_names_whose_canonical_wire_cannot_be_decoded(owner): + with pytest.raises(ValueError): + RepositoryName(owner, "repo") + + +def test_identity_uses_owned_components_not_rendered_text(monkeypatch): + first = RepositoryName("org", "first") + second = RepositoryName("org", "second") + monkeypatch.setattr(RepositoryName, "to_wire", lambda self: "display") + assert first != second + assert first == RepositoryName("ORG", "FIRST") + + +def test_repository_config_json_and_saved_todo_have_the_same_identity(commander): + comms, _ = commander + todo = create(commander, "P1") + repo = Repo.model_validate_json('{"checkout":"/unused", "github":"OPENHCSDEV/AGENT-COMMS"}') + assert repo.holds(todo) + assert Repo.model_validate_json(repo.model_dump_json()) == repo + assert comms.goals.todos.list(repo.github) == (todo,) + restored = comms.goals.todos.get(todo.id) + assert restored.repo == repo.github + assert FieldCodec.decode(type(todo), FieldCodec.encode(todo)) == restored + assert not Repo(checkout=repo.checkout, github=RepositoryName("another", "repo")).holds(todo) + + +def test_order_family_owns_which_kinds_a_merge_closes(commander): + comms, _ = commander + todo = create(commander, "P1") + finding = replace(todo, order=UnorderedProblemOrder()) + assert finding.order.closed_by_merge(finding, (), comms.registry.storage_rule, PullRequests((finding,))) == () diff --git a/tests/test_stale_order_gates.py b/tests/test_stale_order_gates.py index 0e053ea2a1..d4918ac500 100644 --- a/tests/test_stale_order_gates.py +++ b/tests/test_stale_order_gates.py @@ -118,7 +118,7 @@ def publish(order, current, checkout, title, text): assert seen == [('P287', repo, todo.order.pull_request_body(body.read_text()))] subprocess.run(['git', '-C', str(repo), 'remote', 'set-url', 'origin', 'https://github.com/another/project.git'], check=True) - with pytest.raises(SystemExit, match=f'not {todo.repo}'): + with pytest.raises(SystemExit, match=f'not {todo.repo.to_wire()}'): OrderVerb.parse(['pr', 'P287', '--title', 'P287', '--body-file', str(body)], '').apply(context) assert len(seen) == 1 # A different repository must never reach publication. diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index 14858627b5..26f099f238 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -1,3 +1,4 @@ +from agent_comms.repository_name import RepositoryName import importlib import json import os @@ -405,7 +406,7 @@ def git(*args): # Read the captured tracking ref even when the remote is unavailable. git("remote", "set-url", "origin", str(tmp_path / "absent-remote")) config = CommanderConfig.read_root(comms.root).model_copy(update={"repos": ( - Repo(checkout=checkout, github="fleet/growth"),)}) + Repo(checkout=checkout, github=RepositoryName.from_wire("fleet/growth")),)}) config.write(tmp_path / 'growth-home') actual_git, commands = Repo.git, [] diff --git a/tests/test_todos.py b/tests/test_todos.py index 3199ac231f..b5fdca090f 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -16,6 +16,7 @@ from agent_comms.field_codec import FieldCodec from agent_comms.goals import OrderCheck from agent_comms.recorded_time import RecordedEventTime +from agent_comms.repository_name import RepositoryName from agent_comms.registration import Registration from agent_comms.threads import Thread from agent_comms.todos import ( @@ -59,25 +60,25 @@ def test_create_is_idempotent_and_goal_reference_is_historical(tmp_path: Path) - lead = thread("lead", tmp_path, 1.0) goal = GoalRef("lead", 1.0, "goal-1") original = store.create( - "todo-1", "OpenHCSDev/agent-comms", "Fix the sidebar", creator=lead, goal=goal, + "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, goal=goal, owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) - assert original.repo == "openhcsdev/agent-comms" + assert original.repo == RepositoryName.from_wire("OPENHCSDEV/agent-comms") assert original.revision == 1 assert original.assignment == Assignment("lead", 1.0, "lead", 1.0, "initial") assert ( store.create( - "todo-1", "openhcsdev/agent-comms", "Fix the sidebar", creator=lead, goal=goal, + "todo-1", RepositoryName.from_wire("openhcsdev/agent-comms"), "Fix the sidebar", creator=lead, goal=goal, owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) == original ) with pytest.raises(TodoConflict): store.create( - "todo-1", "openhcsdev/agent-comms", "A different task", creator=lead, goal=goal, + "todo-1", RepositoryName.from_wire("openhcsdev/agent-comms"), "A different task", creator=lead, goal=goal, owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) - assert TodoStore(tmp_path).list("openhcsdev/agent-comms") == (original,) + assert TodoStore(tmp_path).list(RepositoryName.from_wire("openhcsdev/agent-comms")) == (original,) newer = thread("lead", tmp_path, 2.0) assert store.get(original.id).goal == goal with pytest.raises(TodoConflict): @@ -88,7 +89,7 @@ def test_create_is_idempotent_and_goal_reference_is_historical(tmp_path: Path) - def test_commander_reads_decode_only_open_todos(tmp_path, monkeypatch): store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) - first = store.create("P1", "org/repo", "Open order", creator=lead, owner=lead, + first = store.create("P1", RepositoryName.from_wire("org/repo"), "Open order", creator=lead, owner=lead, generation="one", goal=GoalRef("lead", 1.0, "request"), deadline=RecordedEventTime(1000), order=order(tmp_path)) second = replace(first, id="P2", state=BlockedTodoState) @@ -146,7 +147,7 @@ def test_two_processes_assign_one_todo_before_any_worker_is_started(tmp_path: Pa store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) store.create( - "todo-1", "OpenHCSDev/agent-comms", "One implementation", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "One implementation", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) with Manager() as manager, ProcessPoolExecutor(max_workers=2) as pool: @@ -162,22 +163,22 @@ def test_two_processes_assign_one_todo_before_any_worker_is_started(tmp_path: Pa persisted = TodoStore(tmp_path).get("todo-1") assert persisted.assignment.owner == observed[0][1] assert persisted.revision == 2 - assert len(store.list("OpenHCSDev/agent-comms")) == 1 + assert len(store.list(RepositoryName.from_wire("OpenHCSDev/agent-comms"))) == 1 def test_validation_rejects_ambiguous_identity_and_bad_state(tmp_path: Path) -> None: store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) for index, repo in enumerate(("org/repo/branch", "../repo", "org/", "org/repo name")): - with pytest.raises(TodoError): + with pytest.raises(ValueError): store.create( - f"invalid-{index}", repo, "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + f"invalid-{index}", RepositoryName.from_wire(repo), "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) with pytest.raises(TodoError): GoalRef("lead", float("nan"), "goal-1") store.create( - "todo-1", "org/repo", "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "todo-1", RepositoryName.from_wire("org/repo"), "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path), ) with pytest.raises(ValueError): @@ -188,7 +189,7 @@ def test_merged_order_keeps_its_checked_fields(tmp_path: Path) -> None: store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) todo = store.create( - "todo-1", "OpenHCSDev/agent-comms", "Fix the sidebar", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path)) merged = replace(todo, order=todo.order.checked().merged("2b9c1f1ab")) cells = merged.order.board_cells(merged) diff --git a/tests/test_tool_replies.py b/tests/test_tool_replies.py index 05dbf355df..40d277e9d6 100644 --- a/tests/test_tool_replies.py +++ b/tests/test_tool_replies.py @@ -1,4 +1,5 @@ """Every declared tool's reply is the encoding of its declared result record.""" +from agent_comms.repository_name import RepositoryName import os from dataclasses import replace from typing import get_type_hints @@ -82,7 +83,7 @@ def scenario(comms: Comms, project, monkeypatch): yield CommsInboxTool, {"thread": "worker", "standby": {"goal_id": "ongoing", "wait_for": ["lead"]}}, worker yield CommsTodoCreateTool, {"owner": "worker", "order": FieldCodec.encode(CreateProblemGoalAction( problem="P1", order=order, request="request", deadline=RecordedEventTime(1000), - commander="lead", repo="OpenHCSDev/agent-comms", text="Fix P1"))}, lead + commander="lead", repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Fix P1"))}, lead yield CommsTodoListTool, {"thread": "lead"}, lead yield CommsUnjudgedTool, {"thread": "lead"}, lead yield CommsCorrectWordingTool, {"unit": "worker", "message_id": "m1", @@ -115,7 +116,7 @@ def scenario(comms: Comms, project, monkeypatch): "artifact": str(comms.root / "installed-evidence")}, lead yield CommsTodoCreateTool, {"owner": "worker", "order": FieldCodec.encode(CreateProblemGoalAction( problem="P2", order=replace(order, id="order-P2"), request="request", deadline=RecordedEventTime(4000), - commander="lead", repo="OpenHCSDev/agent-comms", text="Fix P2"))}, lead + commander="lead", repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Fix P2"))}, lead yield CommsFireTool, {"unit": "worker", "reason": "The unit broke the release path twice and missed both of its deadlines."}, lead yield CommsTurnsTool, {"thread": "worker"}, lead diff --git a/tests/test_work_dispatch.py b/tests/test_work_dispatch.py index 33008f549e..041513ddd1 100644 --- a/tests/test_work_dispatch.py +++ b/tests/test_work_dispatch.py @@ -1,4 +1,5 @@ """State-level sanity checks, not the required native startup/first-reply proof.""" +from agent_comms.repository_name import RepositoryName from dataclasses import replace @@ -61,7 +62,7 @@ def test_invalid_request_refuses_dispatch_before_allocating_anything(commander): comms, _ = commander checked = order(comms.root, "full-phase-order") action = CreateProblemGoalAction( - problem="P1", repo="OpenHCSDev/agent-comms", text="Repair the full phase", + problem="P1", repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Repair the full phase", order=checked, request="not-the-active-request", commander="lead", deadline=RecordedEventTime(1000), ) diff --git a/tests/verify_connector_tools.py b/tests/verify_connector_tools.py index 65cad711ec..913324ef02 100644 --- a/tests/verify_connector_tools.py +++ b/tests/verify_connector_tools.py @@ -6,6 +6,7 @@ """ from __future__ import annotations +from agent_comms.repository_name import RepositoryName import argparse import json @@ -82,7 +83,7 @@ def arguments(comms: Comms, name: str, project: Path, checkout: Path) -> dict: request="request", deadline=RecordedEventTime(time.time() + 600), commander="claude-ai", - repo="OpenHCSDev/agent-comms", + repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Measure the connector order.", ) order_tools = { 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()