Integrate shell decoding and assigned-scratch enforcement

src/agent_comms/plan_phases.py

@@ -230,7 +230,7 @@ class RecordPlanePhase(CommanderNativePhase):
 class ReviewQueuePhase(CommanderNativePhase):
     """A verdict per PR head gates merging through main's required review status.
 
-    The owner's admin bypass stays; its hook demands a recorded verdict on that head.
+    QueueBypass declares each bypass's pin and recorded-verdict obligations.
     No GitHub merge queue is required.
     """
 

src/agent_comms/shell_effects.py

@@ -1,3 +1,3 @@
 """Shell command effects, decoded below storage enforcement and order consumers."""
 from __future__ import annotations
 
@@ -45,8 +45,8 @@
     def matches(args: Sequence[str], prefixes: Sequence[tuple[str, ...]]) -> bool:
         return any(tuple(args[:len(prefix)]) == prefix for prefix in prefixes)
 
-    def invocation(self, decoded: ShellCommand, cwd: Path) -> ShellInvocation:
-        return ShellInvocation(decoded.words, cwd)
+    def following_directory(self, args: list[str], cwd: Path) -> Path:
+        return cwd
 
     def effects(self, args: list[str], cwd: Path) -> tuple[StorageEffect, ...]:
         """Unknown programs have no declared effects; capabilities compose through super()."""
@@ -158,25 +158,29 @@
     @classmethod
     def parse(cls, command: str, cwd: Path) -> Iterator[ShellInvocation]:
         try:
-            decoded_commands = tuple(ShellCommand.split(command))
+            yield from cls.at_directory(ShellCommand.parsed(command), cwd)
         except ValueError as error:
             raise StorageDenied(f'StorageRule: the shell command cannot be decoded ({error}); '
                                 'no tool was started') from error
-        for decoded in decoded_commands:
-            invocation = ShellProgram.named(decoded.program).invocation(decoded, cwd)
-            cwd = invocation.cwd
-            yield invocation
+
+    @classmethod
+    def at_directory(cls, commands: Iterator[ShellCommand], cwd: Path) -> Iterator[ShellInvocation]:
+        for decoded in commands:
+            yield cls(decoded.words, cwd, redirections=decoded.redirections)
+            # Child execution inherits this directory; its changes die with that child.
+            yield from cls.at_directory(decoded.nested(), cwd)
+            cwd = ShellProgram.named(decoded.program).following_directory(decoded.args, cwd)
 
 
 class ChangeDirectoryProgram(ShellProgram, names=frozenset({'cd'})):
-    def invocation(self, decoded: ShellCommand, cwd: Path) -> ShellInvocation:
+    def following_directory(self, args: list[str], cwd: Path) -> Path:
         try:
-            _, targets = getopt.getopt(decoded.words[1:], '')
+            _, targets = getopt.getopt(args, '')
         except getopt.GetoptError as error:
             raise StorageDenied(f'StorageRule: unknown cd option: {error}') from error
         if targets:
             cwd = physical_path(targets[0], cwd)
-        return ShellInvocation(decoded.words, cwd)
+        return cwd
 
 
 @dataclass(frozen=True)
@@ -238,15 +242,22 @@
 
 
 class WritesTargets(ShellProgram):
-    targets: ClassVar[slice] = slice(None)
-    minimum: ClassVar[int] = 1
+    """The declaring CLI owns option arity; only decoded operands become paths."""
+
+    short_options: ClassVar[str] = ''
+    long_options: ClassVar[tuple[str, ...]] = ('help', 'version')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        return tuple(arguments)
 
     def effects(self, args: list[str], cwd: Path) -> tuple[StorageEffect, ...]:
-        paths = [arg for arg in args if not arg.startswith('-')]
-        selected = tuple(
-            WriteStorageEffect(physical_path(path, cwd)) for path in paths[self.targets]
-        ) if len(paths) >= self.minimum else ()
-        return super().effects(args, cwd) + selected
+        try:
+            options, arguments = getopt.gnu_getopt(args, self.short_options, self.long_options)
+        except getopt.GetoptError as error:
+            raise StorageDenied(f'StorageRule: incomplete or unknown file-program option: {error}') from error
+        paths = () if any(option in ('--help', '--version') for option, _ in options) else self.destinations(options, arguments)
+        return super().effects(args, cwd) + tuple(
+            WriteStorageEffect(physical_path(path, cwd)) for path in paths)
 
 
 class WritesWorkingTree(ShellProgram):
@@ -348,22 +359,121 @@
     prefixes = (('record',),)
 
 
-class FileWriterProgram(WritesTargets, names=frozenset({'mkdir', 'touch', 'tee'})):
-    pass
+class MakeDirectoryProgram(WritesTargets, names=frozenset({'mkdir'})):
+    short_options = 'm:pvZ'
+    long_options = (*WritesTargets.long_options, 'mode=', 'parents', 'verbose', 'context=?')
+
+
+class TouchProgram(WritesTargets, names=frozenset({'touch'})):
+    short_options = 'acd:fhmr:t:'
+    long_options = (*WritesTargets.long_options, 'no-create', 'date=', 'no-dereference', 'reference=', 'time=')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        # '-' names the selected stdout descriptor, not a file named '-'.
+        return tuple(path for path in arguments if path != '-')
+
+
+class TeeProgram(WritesTargets, names=frozenset({'tee'})):
+    short_options = 'aip'
+    long_options = (*WritesTargets.long_options, 'append', 'ignore-interrupts', 'output-error=?')
+
+
+class CopyProgram(WritesTargets):
+    """Copy-like CLIs share the final destination operand, not their option meanings."""
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        return tuple(arguments[-1:]) if len(arguments) >= 2 else ()
+
+
+class TargetDirectoryCopyProgram(CopyProgram):
+    short_options = 't:S:'
+    long_options = (*CopyProgram.long_options, 'target-directory=', 'suffix=')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        directories = tuple(value for option, value in options if option in ('-t', '--target-directory'))
+        return directories or super().destinations(options, arguments)
+
+
+class CopyFilesProgram(TargetDirectoryCopyProgram, names=frozenset({'cp'})):
+    short_options = TargetDirectoryCopyProgram.short_options + 'abdfiHLPlnpRrsTuvxZ'
+    long_options = (*TargetDirectoryCopyProgram.long_options, 'archive', 'attributes-only', 'backup=?', 'copy-contents',
+                    'no-dereference', 'debug', 'force', 'interactive', 'dereference', 'keep-directory-symlink',
+                    'link', 'no-clobber', 'preserve=?', 'no-preserve=', 'parents', 'recursive', 'reflink=?',
+                    'remove-destination', 'sparse=', 'strip-trailing-slashes', 'symbolic-link',
+                    'no-target-directory', 'update=?', 'verbose', 'one-file-system', 'context=?')
+
+
+class MoveProgram(TargetDirectoryCopyProgram, names=frozenset({'mv'})):
+    short_options = TargetDirectoryCopyProgram.short_options + 'bfinTuvZ'
+    long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'debug', 'exchange', 'force', 'interactive',
+                    'no-clobber', 'no-copy', 'strip-trailing-slashes', 'no-target-directory',
+                    'update=?', 'verbose', 'context')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        # Moving also removes source directory entries; destination-only admission hid that write.
+        return tuple(dict.fromkeys((*super().destinations(options, arguments), *arguments)))
+
+
+class InstallFilesProgram(TargetDirectoryCopyProgram, names=frozenset({'install'})):
+    short_options = TargetDirectoryCopyProgram.short_options + 'bcCdDg:m:o:psTvZ'
+    long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'compare', 'directory', 'debug', 'group=',
+                    'mode=', 'owner=', 'preserve-timestamps', 'strip', 'strip-program=',
+                    'no-target-directory', 'verbose', 'preserve-context', 'context=?')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        if any(option in ('-d', '--directory') for option, _ in options):
+            return tuple(arguments)
+        return super().destinations(options, arguments)
+
+
+class LinkProgram(TargetDirectoryCopyProgram, names=frozenset({'ln'})):
+    short_options = TargetDirectoryCopyProgram.short_options + 'bdFfiLnPrsTv'
+    long_options = (*TargetDirectoryCopyProgram.long_options, 'backup=?', 'directory', 'force', 'interactive', 'logical',
+                    'no-dereference', 'physical', 'relative', 'symbolic', 'no-target-directory', 'verbose')
+
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        selected = super().destinations(options, arguments)
+        return selected or (('.',) if len(arguments) == 1 else ())
+
 
+class RsyncProgram(CopyProgram, names=frozenset({'rsync'})):
+    # Remote/native expansion is not inferred. Unknown options are explicit refusals.
+    short_options = '0468ACDEFGHIJKLNOPRSUVWXM:B:T:e:f:abcdghiklmnopqrstuvxyz@:'
+    long_options = (*WritesTargets.long_options, 'verbose', 'quiet', 'archive', 'recursive', 'relative',
+                    'checksum', 'links', 'copy-links', 'perms', 'owner', 'group', 'times', 'dry-run',
+                    'delete', 'delete-before', 'delete-during', 'delete-after', 'progress', 'stats',
+                    'human-readable', 'itemize-changes', 'compress', 'one-file-system', 'inplace',
+                    'partial', 'numeric-ids', 'ignore-existing', 'size-only', 'remove-source-files',
+                    'info=', 'debug=', 'stderr=', 'suffix=', 'backup-dir=', 'chmod=', 'chown=',
+                    'block-size=', 'rsh=', 'rsync-path=', 'timeout=', 'contimeout=', 'modify-window=',
+                    'temp-dir=', 'partial-dir=', 'compare-dest=', 'copy-dest=', 'link-dest=',
+                    'filter=', 'exclude=', 'exclude-from=', 'include=', 'include-from=', 'files-from=',
+                    'out-format=', 'log-file=', 'log-file-format=', 'bwlimit=', 'max-size=', 'min-size=',
+                    'write-batch=', 'only-write-batch=', 'read-batch=', 'remote-option=')
+    receiver_writes: ClassVar[frozenset[str]] = frozenset({
+        '-T', '--temp-dir', '--partial-dir', '--backup-dir'})
+    auxiliary_writes: ClassVar[frozenset[str]] = receiver_writes | frozenset({
+        '--log-file', '--write-batch', '--only-write-batch'})
 
-class CopyProgram(WritesTargets, names=frozenset({'cp', 'mv', 'rsync', 'install', 'ln'})):
-    targets = slice(-1, None)
-    minimum = 2
+    def destinations(self, options: list[tuple[str, str]], arguments: list[str]) -> tuple[str, ...]:
+        paths = super().destinations(options, arguments)
+        for option, value in options:
+            if option in self.receiver_writes and not Path(value).is_absolute():
+                raise StorageDenied('StorageRule: use an absolute rsync receiver-side destination')
+        paths += tuple(value for option, value in options if option in self.auxiliary_writes)
+        if any(option == '--remove-source-files' for option, _ in options):
+            paths += tuple(arguments[:-1])
+        if any(':' in path for path in paths):
+            raise StorageDenied('StorageRule: remote rsync destinations require a separately admitted transport')
+        return tuple(dict.fromkeys(paths))
 
 
 def shell_storage_effects(command: str, cwd: Path) -> tuple[StorageEffect, ...]:
     """Decode known shell operations, not arbitrary interpreter programs."""
     effects: list[StorageEffect] = []
     for invocation in ShellInvocation.parse(command, cwd):
-        args = invocation.args
-        for index, word in enumerate(args[:-1]):
-            if word in ('>', '>>'):
-                effects.append(WriteStorageEffect(physical_path(args[index + 1], invocation.cwd)))
-        effects.extend(ShellProgram.named(invocation.program).effects(args, invocation.cwd))
+        for redirection in invocation.redirections:
+            effects.extend(WriteStorageEffect(physical_path(path, invocation.cwd))
+                           for path in redirection.written_paths())
+        effects.extend(ShellProgram.named(invocation.program).effects(invocation.args, invocation.cwd))
     return tuple(effects)

src/agent_comms/storage_rules.py

@@ -689,14 +689,17 @@ def require_write(self, path: Path) -> None:
         owner, = owners
         resolved = path.resolve()
         sources = Checkout.git_roots(resolved, *resolved.parents)
+        scratch = self.rule.roots_of(owner).scratch.resolve()
+        # The registry's scratch allocation is more specific than an enclosing
+        # source tree. A repository rooted in that allocation is still source.
+        if resolved.is_relative_to(scratch) and not any(
+                root.is_relative_to(scratch) for root in sources):
+            return
         store = self.comms.goals.todos
         allowed = frozenset(boundary for root in sources
                             for allocation in CheckoutAllocation.for_path(self.rule, owner, root)
                             for boundary in allocation.edit_boundary(self.rule, owner, root, store, self.reads))
-        # A thread registered at a directory holding its scratch (unit-fieldplan at $HOME) still
-        # owns that scratch: the rule gate's audit lives there, and refusing it deadlocked the unit.
-        in_project = (resolved.is_relative_to(owner.project_path.resolve())
-                      and not resolved.is_relative_to(self.rule.roots_of(owner).scratch.resolve()))
+        in_project = resolved.is_relative_to(owner.project_path.resolve())
         if (sources or path.is_relative_to(self.rule.checkout_base) or in_project) and not any(
                 resolved.is_relative_to(root) for root in allowed):
             self.rule.deny(f'{path} is a read/launch source, not allocated editable work of {owner.name}.')

src/agent_comms_hooks/bounded_waits.py

@@ -1,3 +1,3 @@
 """A Bash wait loop must end on its own: it runs as `timeout N bash -c '...'`, and its
 condition never uses `pgrep -f`, whose pattern also matches the waiter's own command line
 and every other waiter naming the same task, so such loops keep each other alive forever.
@@ -18,7 +18,7 @@
 
 from .shell_commands import (
     ShellCommand,
-    shell_commands,
+    TimeoutRunner,
 )
 from .watch import SessionHookProgram
 
@@ -31,9 +31,7 @@
     """`timeout [OPTION]... DURATION COMMAND`: whether it bounds the command it runs, and
     that command. Any other command runs itself, unbounded."""
 
-    program: ClassVar[str] = "timeout"
-    valued: ClassVar[frozenset[str]] = frozenset(("-k", "-s", "--kill-after", "--signal"))
-    """Options whose value is the next word."""
+    program: ClassVar[str] = TimeoutRunner.names[0]
     duration: ClassVar[re.Pattern[str]] = re.compile(r"(\d+(?:\.\d*)?)([smhd]?)")
     unit: ClassVar[dict[str, int]] = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}
     max_seconds: ClassVar[int] = 3600
@@ -45,9 +43,7 @@
     def around(cls, command: ShellCommand) -> Timeout:
         if command.program != cls.program:
             return cls(False, command.words)
-        rest = command.words[1:]
-        while rest[:1] and rest[0].startswith("-"):
-            rest = rest[2:] if rest[0] in cls.valued else rest[1:]
+        rest = tuple(TimeoutRunner.arguments(command.args))
         limit = cls.duration.fullmatch(rest[0]) if rest else None
         seconds = float(limit.group(1)) * cls.unit[limit.group(2)] if limit else 0
         return cls(0 < seconds <= cls.max_seconds, rest[1:])
@@ -93,39 +89,28 @@
     """The word that ends a loop's condition."""
     openers: ClassVar[frozenset[str]] = frozenset((*body, "then", "!", "{"))
     """Words after which the next word is again a command."""
-    shell: ClassVar[re.Pattern[str]] = re.compile(r"(?:ba)?sh")
-    script_flag: ClassVar[re.Pattern[str]] = re.compile(r"-\w*c")
-
     @classmethod
     def commands(cls, segment: Words) -> Iterator[ShellCommand]:
-        """What one segment runs: its command, or the command its `timeout` runs."""
+        """Execution descendants derive from the same runner/group declarations."""
         for command in ShellCommand.of(dropwhile(cls.openers.__contains__, segment)):
-            yield from Timeout.around(command).commands()
+            yield from command.run_commands()
 
     def waits(self) -> Iterator[Wait]:
-        """Every loop this command runs, at command position or in a script a shell runs with
-        -c. Only such a script can be bounded: `timeout` runs programs, not shell keywords."""
-        segments = tuple(map(tuple, shell_commands(self.command)))
-        for index, segment in enumerate(segments):
-            for command in ShellCommand.of(dropwhile(self.openers.__contains__, segment)):
-                timeout = Timeout.around(command)
-                for run in timeout.commands():
-                    match (run.program, *run.args):
-                        case (keyword, *condition) if keyword in self.loops:
-                            yield Wait(
-                                False,
-                                (
-                                    tuple(condition),
-                                    *takewhile(
-                                        lambda later: later[:1] != self.body, segments[index + 1 :]
-                                    ),
-                                ),
-                            )
-                        case (program, flag, script, *_) if self.shell.fullmatch(
-                            program
-                        ) and self.script_flag.fullmatch(flag):
-                            for wait in WaitCommand(script).waits():
-                                yield replace(wait, bounded=wait.bounded or timeout.bounded)
+        return self.in_scope(tuple(ShellCommand.parsed(self.command)), False)
+
+    @classmethod
+    def in_scope(cls, commands: tuple[ShellCommand, ...], bounded: bool) -> Iterator[Wait]:
+        """Shell source owns control words; executable argv does not become shell syntax."""
+        segments = tuple(command.source_words for command in commands)
+        for index, decoded in enumerate(commands):
+            command = replace(decoded, words=tuple(dropwhile(cls.openers.__contains__, decoded.words)))
+            timeout = Timeout.around(command)
+            for source in command.source_commands():
+                for run in Timeout.around(source).commands():
+                    if run.program in cls.loops:
+                        yield Wait(bounded, (tuple(run.args), *takewhile(
+                            lambda later: later[:1] != cls.body, segments[index + 1:])))
+            yield from cls.in_scope(tuple(command.nested()), bounded or timeout.bounded)
 
     def unbounded_wait(self) -> tuple[str, ...]:
         """Why this command's wait loop could run forever; empty when it holds no loop."""

src/agent_comms_hooks/queue_bypass.py

@@ -144,7 +144,7 @@ def matching(cls, command: ShellCommand) -> tuple[QueueBypass, ...]:
 
 
 class AdminMerge(QueueBypass):
-    """GitHub's head precondition and Core's verdict name the same commit."""
+    """A pinned admin invocation supplies its review without a saved Core verdict."""
 
     reason = "`gh pr merge --admin` bypasses required checks."
     prefix = ("pr", "merge")