Fix six owners behind 18 red tests on main

src/agent_comms/goals.py

@@ -137,9 +137,9 @@ def project(self, registry, owner, goal, source):
         contacts = tuple(
             GoalDerivedContact(
                 owner.name,
-                owner.created_at,
+                owner.incarnation.wire_birth,
                 executor.name,
-                executor.created_at,
+                executor.incarnation.wire_birth,
                 goal.id,
                 source.text_revision,
             )

src/agent_comms/installer/backup.py

@@ -340,10 +340,15 @@ def copy_tree(
         omitted.update(member for path, owner in owners.items()
                        for member in owner.files(path) if member != path)
 
+        def passes(path: Path) -> bool:
+            """Directories and links pass through; a file passes only with its owner, so one
+            written after the owners were read, or a socket, is never copied."""
+            listed = path in owners or stat.S_ISDIR(path.lstat().st_mode) or path.is_symlink()
+            return listed and path not in omitted
+
         def transient(directory: str, names: list[str]) -> set[str]:
             return {path.name for path in (Path(directory) / name for name in names)
-                    if path in omitted or LockedStore.is_staging(path)
-                    or stat.S_ISSOCK(path.lstat().st_mode)}
+                    if LockedStore.is_staging(path) or not passes(path)}
 
         def copy_file(original: str, destination: str) -> Path:
             return owners[Path(original)].copy_file(original, destination)

src/agent_comms/storage_rules.py

@@ -10,7 +10,7 @@
 import time
 from abc import abstractmethod
 from collections.abc import Iterable, Iterator, Sequence
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from enum import StrEnum
 from functools import cached_property
 from pathlib import Path
@@ -84,9 +84,11 @@ class StorageRule(RoomAllocation, DurableFile):
     shown: ClassVar[int] = 5  # holders named in each list
     scratch_cap_gib: float = 3.0  # one unit's scratch together with its checkouts
     max_checkouts: int = 2
-    volume: StoragePath = Path.home()
-    scratch_base: StoragePath = Path.home() / '.cache' / 'agent-scratch'
-    checkout_base: StoragePath = Path.home() / 'wt'
+    # Home paths are read when the rule is built, never frozen at import.
+    volume: StoragePath = field(default_factory=Path.home)
+    scratch_base: StoragePath = field(
+        default_factory=lambda: Path.home() / '.cache' / 'agent-scratch')
+    checkout_base: StoragePath = field(default_factory=lambda: Path.home() / 'wt')
 
     def __post_init__(self):
         limits = (self.floor_gib, self.cleanup_gib, self.scratch_cap_gib)

src/agent_comms/todos.py

@@ -2181,7 +2181,7 @@ def require_send(self, context: CommanderContext, document: OrderFile, unit: str
         problems.extend(self.assignment_refusals(
             context, document, unit, group, listed, target, traced))
         problems.extend(self.source_refusals(context, document, traced))
-        problems.extend(self.check_refusals(context.worktrees))
+        problems.extend(self.check_refusals(context.comms.registry.storage_rule.checkout_base))
         problems.extend(reassign.refusals(holders))
         if problems:
             sys.exit('Refused order:\n  ' + '\n  '.join(problems))
@@ -3912,16 +3912,15 @@ def require_assignment_in(self, db: sqlite3.Connection, problem: str,
         """Creation and reassignment share this decision under registry then Todo locks."""
         self.require_unit_in(db, unit, assigner)
         snapshot = RegistryStore.at(self.root).read_snapshot_unlocked()
-        owner = snapshot.by_lookup[unit.require_recorded().lookup]
-        worktree = owner.project_path
+        worktrees = {owner.project_path for owner in snapshot.threads_for(unit.require_recorded())}
         for held in self._open(db, "assignment IS NOT NULL", ()):
             previous = held.held().owner_identity
             if held.id == problem or previous == unit:
                 continue
             for other in snapshot.threads_for(previous.require_recorded()):
-                if worktree == other.project_path:
+                if other.project_path in worktrees:
                     raise SharedWorktreeRefusal(
-                        problem, held.id, owner.incarnation, other.incarnation, worktree
+                        problem, held.id, unit, other.incarnation, other.project_path
                     )
 
     def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord:

tests/test_commander_forks.py

@@ -51,9 +51,13 @@ def fork_bus(tmp_path):
     comms = wire(tmp_path / 'saved-bus')
     Bus.install(Bus.path_in(comms.root))
     comms.channels.create_tag('comms')
-    for name in ('lead', 'worker', 'other', 'retained'):
+    # Two order holders never share one worktree, so each unit has its own checkout.
+    units = tuple((name, tmp_path / f'{name}-checkout') for name in ('worker', 'other', 'retained'))
+    for name, worktree in units:
+        source.git(source.checkout, 'worktree', 'add', '-q', '-b', f'{name}-work', str(worktree))
+    for name, worktree in (('lead', source.checkout), *units):
         comms.registry.declare(comms.registry.new_thread(
-            name, frozenset({'comms'}), str(source.checkout),
+            name, frozenset({'comms'}), str(worktree),
             model=DeclaredModel('anthropic/claude-opus-4-6'), execution=ExternalThreadExecution))
     comms.goals.update_goal('lead', SetGoalAction(
         text='Repair the commander workflow', goal_id='request'))

tests/test_connector.py

@@ -69,13 +69,14 @@ def test_real_connector_refusals_and_replaced_source(tmp_path, source_programs):
     """The actual serve entrypoint, curl transport and production maintenance thread."""
     import os
     import subprocess
-    import sys
 
     project = Path(__file__).resolve().parents[1]
     environment = {**os.environ, "PYTHONPATH": str(project / "src"),
                    "PATH": os.pathsep.join((str(source_programs.prefix / "bin"), os.environ["PATH"]))}
+    # The proof creates the root it serves, so it runs under the serving runtime.
     proof = subprocess.run(
-        [sys.executable, "-X", "faulthandler", str(project / "tests/verify_connector_live.py"),
+        [str(source_programs.prefix / "bin/python"), "-X", "faulthandler",
+         str(project / "tests/verify_connector_live.py"),
          str(tmp_path / "live-connector")],
         env=environment, capture_output=True, text=True, timeout=60,
     )

tests/test_holder_standing_rules.py

@@ -76,9 +76,13 @@ def held_bus(comms, tmp_path, source_repository, configured_repository):
     source = source_repository
     scope = source.scope()
     Bus.install(Bus.path_in(comms.root))
+    # Two order holders never share one worktree, so the outsider has its own checkout.
+    outsider = tmp_path / 'outsider-checkout'
+    source.git(source.checkout, 'worktree', 'add', '-q', '-b', 'outsider-work', str(outsider))
     declare_test_bus(comms, channels=('comms',), threads=tuple(
-        comms.registry.new_thread(name=name, tags=frozenset({'comms'}), worktree=str(source.checkout))
-        for name in ('lead', 'worker', 'outsider')
+        comms.registry.new_thread(name=name, tags=frozenset({'comms'}), worktree=str(worktree))
+        for name, worktree in (('lead', source.checkout), ('worker', source.checkout),
+                               ('outsider', outsider))
     ))
     request = comms.goals.todos.add_request(
         comms.registry.require('lead').incarnation, 'Repair the held work').id

tests/test_todos.py

@@ -270,7 +270,7 @@ def test_merged_order_keeps_its_checked_fields(tmp_path: Path, source_repository
 
 
 @pytest.fixture
-def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, configured_repository):
+def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, durable_home):
     """One cited source, its real bare remote and installed history; only GitHub is synthetic."""
     graph = next(graph for graph in proof_build_checkouts
                  if graph.repository.github == CorePackage.repository)
@@ -311,7 +311,9 @@ def pr_commander(tmp_path, monkeypatch, proof_build_checkouts, configured_reposi
     rule = StorageRule(checkout_base=source.checkout.parent)
     (tmp_path / "storage.json").write_text(json.dumps(FieldCodec.encode(rule)))
     Registration(tmp_path / "registry.json").register(lead)
-    configured_repository(wire(tmp_path), source)
+    # The source carries the copied plans; a second checkout of its repository would
+    # be compared against the source's installed commit, which it never holds.
+    Commander.on_root('lead', (source,), tmp_path)
     bus = wire(tmp_path)
     bus.goals.todos.add_request(lead.incarnation, "Repair the copied order workflow")
     return Commander.read_root(tmp_path)

tests/test_unowned_checkout_work.py

@@ -130,9 +130,12 @@ def checkout_bus(tmp_path: Path, configured_repository, installed_runtime) -> Co
     Bus.install(Bus.path_in(root))
     c = wire(root)
     Bus.install(Bus.path_in(c.root))
-    for name in ('lead', 'worker', 'idle'):
+    # Two order holders never share one worktree, so the unit taking the work has its own.
+    spare = tmp_path / 'idle-project'
+    git(home, 'worktree', 'add', '-q', '-b', 'idle-home', str(spare))
+    for name, worktree in (('lead', home), ('worker', home), ('idle', spare)):
         c.registry.declare(c.registry.new_thread(
-            name, frozenset(), str(home), execution=ExternalThreadExecution))
+            name, frozenset(), str(worktree), execution=ExternalThreadExecution))
     request = c.goals.todos.add_request(
         c.registry.require('lead').incarnation, 'Keep changes with their author').id
     lead, worker = c.registry.require('lead'), c.registry.require('worker')