feat(docker): a control-plane image for agent_sys, carrying no engine

agent_sys/agent/backends/claude_sdk.py

@@ -466,9 +466,45 @@ def _deliver(self, message: str) -> None:
         self._await(self._client.query(message))
 
     def _terminate(self) -> None:
-        if self._connected:
-            self._await(self._client.disconnect())
+        """Disconnect the SDK and deterministically dispose of its private loop.
+
+        A successful submission deliberately leaves the client connected: the
+        completeness gate may request another submission on the same session.
+        `stop()` is the owner boundary where that reuse has ended. Merely
+        dropping the loop here leaves the SDK's `Query._read_messages` task and
+        subprocess transport alive; Python then destroys them after the loop is
+        closed and emits `Task was destroyed but it is pending`.
+
+        Idempotence matters because scheduler stop and process shutdown can both
+        reach the same executor.
+        """
+        with self._loop_lock:
+            if self._loop.is_closed():
+                self._connected = False
+                return
+            try:
+                self._loop.run_until_complete(self._shutdown())
+            finally:
+                self._loop.close()
+
+    async def _shutdown(self) -> None:
+        """Close the SDK first, then cancel anything it left on this loop."""
+        try:
+            if self._connected:
+                await self._client.disconnect()
+        finally:
             self._connected = False
+            current = asyncio.current_task()
+            pending = [
+                task
+                for task in asyncio.all_tasks(self._loop)
+                if task is not current and not task.done()
+            ]
+            for task in pending:
+                task.cancel()
+            if pending:
+                await asyncio.gather(*pending, return_exceptions=True)
+            await self._loop.shutdown_asyncgens()
 
     # ---- level 2 ---------------------------------------------------------- #
 

agent_sys/agent/runner.py

@@ -182,6 +182,36 @@ def stop(self, task_id: TaskId, on_stopped: Callable[[TaskId], None]) -> None:
             self._attempts.pop(task_id, None)
         on_stopped(task_id)
 
+    def shutdown(self) -> None:
+        """Release every attempt when the process-level run is over.
+
+        Attempts outlive their worker thread on purpose: a monitor can wake a
+        parked task or submit to the same executor again while the run exists.
+        Once all monitors have stopped and the CLI has produced its report,
+        there is no remaining owner for those executors. Keeping them in
+        `_attempts` leaked each Claude SDK reader task and subprocess transport
+        until interpreter teardown.
+
+        The map is emptied before any potentially blocking backend shutdown, so
+        no concurrent observer can acquire an executor whose disposal started.
+        Every attempt is given a chance to stop even when one backend raises;
+        the first error is re-raised after all worker threads have been joined.
+        """
+        with self._lock:
+            attempts = tuple(self._attempts.values())
+            self._attempts.clear()
+
+        first_error: Exception | None = None
+        for attempt in attempts:
+            try:
+                attempt.halt()
+            except Exception as exc:  # cleanup all owners before reporting one failure
+                first_error = first_error or exc
+        for attempt in attempts:
+            attempt.join(HANDOVER_GRACE)
+        if first_error is not None:
+            raise first_error
+
     # ---- the monitor's two entrances, not on the protocol ----------------- #
 
     # `Runner.resume` is gone. `carry_on` subsumed it, `monitor`'s Protocol
@@ -560,9 +590,12 @@ def halt(self) -> None:
         """`Runner.stop`: end the executor and the thread."""
         with self._own:
             self._halted = True
-        if self.executor is not None:
-            self.executor.stop()
-        self._wake.set()
+        try:
+            if self.executor is not None:
+                self.executor.stop()
+        finally:
+            # A backend cleanup error must not leave the attempt parked forever.
+            self._wake.set()
 
     def join(self, timeout: float | None = None) -> None:
         """For a caller that wants the thread settled. Tests, and `demo`."""

agent_sys/cli/environment.py

@@ -45,7 +45,9 @@
     "Layout",
     "LiveHandoffs",
     "build_context",
+    "WORKROOT_ENV_VAR",
     "confinement",
+    "default_root",
     "demo_grants",
     "latest_run",
     "layout_for",
@@ -214,14 +216,48 @@ def __len__(self) -> int:
         return 0 if self._mgr is None else len(self._mgr.all_ids())
 
 
+#: The run root, named outright. `--demo-root` still wins; this is what a
+#: caller who cannot pass an argument sets.
+#:
+#: **It names the root itself, not a base to append to.** `XDG_STATE_HOME` is a
+#: base — the spec says what may go under it and every program appends its own
+#: name — so `agent-sys-demo` is this program's share of a directory it does not
+#: own. That is the wrong shape for the thing this variable exists to control:
+#: the run root has to be **one absolute path that resolves identically here and
+#: on the compute node**, because `remote.sh:require_visible_on_node` asserts
+#: exactly that and the bodies run out of `<run root>/runs/<id>/zones/…/package`
+#: by absolute path. A variable that only moves the parent leaves the operator
+#: composing the real answer in their head, and a container bind mount has to
+#: name the whole path anyway.
+#:
+#: `INFERA_` and not `AGENT_SYS_`: the `AGENT_SYS_*` namespace is what `env_mgr`
+#: *publishes to a body* — `AGENT_SYS_MY_ZONE`, `AGENT_SYS_MY_WORKSPACE`,
+#: `AGENT_SYS_TASK_PACKAGE` — and one of those set by a caller is either ignored
+#: or a collision. A variable read *from* the environment does not belong in a
+#: namespace whose other members are written *to* it.
+WORKROOT_ENV_VAR = "INFERA_AGENT_SYSTEM_WORKROOT"
+
+
 def default_root() -> Path:
-    """`$XDG_STATE_HOME/agent-sys-demo`, or `~/.local/state/...`.
+    """`$INFERA_AGENT_SYSTEM_WORKROOT`, else `$XDG_STATE_HOME/agent-sys-demo`.
+
+    Two sources and one rule: the specific name wins over the generic base. A
+    caller who set neither gets `~/.local/state/agent-sys-demo`.
 
     State rather than cache or data: it is *"state that should persist between
     restarts but is not important enough for the data directory"*, which is what
     a demo run is. A cache directory would be correct until somebody cleared it
     between the interrupt and the resume.
+
+    **An empty or relative value reads as unset.** `XDG_STATE_HOME`'s own
+    specification says exactly that of a base directory, and the reason applies
+    with more force here: a relative run root resolves against whatever `cwd` a
+    body inherited, which is the one thing `<run root>` may not depend on if the
+    compute node is to find the same directory.
     """
+    named = os.environ.get(WORKROOT_ENV_VAR, "").strip()
+    if named and os.path.isabs(named):
+        return Path(named)
     base = os.environ.get("XDG_STATE_HOME") or str(Path.home() / ".local" / "state")
     return Path(base) / "agent-sys-demo"
 

agent_sys/cli/main.py

@@ -173,6 +173,47 @@ def parser() -> argparse.ArgumentParser:
             "ends in seconds regardless; this only bounds one that never stops"
         ),
     )
+
+    # Docker mode: run inside a container managed by env_mgr.
+    run.add_argument(
+        "--docker",
+        action="store_true",
+        help="run inside a Docker container (env_mgr builds and starts it automatically)",
+    )
+    run.add_argument(
+        "--docker-debug",
+        action="store_true",
+        help="start the container and drop into an interactive shell (sleep infinity), no task is run",
+    )
+    run.add_argument(
+        "--docker-image",
+        metavar="IMAGE",
+        default="infera/agent-sys:latest",
+        help="Docker image to use (default: infera/agent-sys:latest)",
+    )
+    run.add_argument(
+        "--docker-name",
+        metavar="NAME",
+        default="agent-sys-container",
+        help="container name (default: agent-sys-container)",
+    )
+    run.add_argument(
+        "--docker-rm",
+        action="store_true",
+        help="remove the container after the run finishes (default: keep running)",
+    )
+    run.add_argument(
+        "--detect-and-copy-host-ssh-config",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="mount host ~/.ssh into the container (default: on)",
+    )
+    run.add_argument(
+        "--detect-and-copy-host-claude-config",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="mount host ~/.claude into the container (default: on)",
+    )
     return top
 
 
@@ -262,11 +303,97 @@ def _show(args: argparse.Namespace, stream: Stream) -> int:
 def _run(args: argparse.Namespace, stream: Stream) -> int:
     if args.clean:
         return _clean(args, stream)
+    if getattr(args, "docker_debug", False):
+        return _docker_debug(args, stream)
+    if getattr(args, "docker", False):
+        return _docker_run(args, stream)
     if args.dry_run:
         return _dry_run(args, stream)
     return _real_run(args, stream)
 
 
+def _docker_debug(args: argparse.Namespace, stream: Stream) -> int:
+    """Start a container with sleep infinity and drop into a shell."""
+    from env_mgr.container import ContainerManager
+
+    mgr = ContainerManager(
+        image=args.docker_image,
+        container_name=args.docker_name,
+        detect_ssh=args.detect_and_copy_host_ssh_config,
+        detect_claude=args.detect_and_copy_host_claude_config,
+    )
+
+    mgr.start()
+
+    name = mgr.container_name
+    stream.emit(
+        EventKind.RUN_COMPLETE,
+        f"container '{name}' is running (sleep infinity). Connect with:\n"
+        f"\n"
+        f"  docker exec -it {name} bash\n",
+        exit_code=OK,
+        ok=True,
+    )
+
+    import subprocess
+    return subprocess.run(
+        [mgr._docker_bin(), "exec", "-it", "-w", "/opt/Infera", name, "bash"],
+    ).returncode
+
+
+def _docker_run(args: argparse.Namespace, stream: Stream) -> int:
+    """Delegate the run to a Docker container managed by env_mgr."""
+    from env_mgr.container import ContainerManager
+
+    mgr = ContainerManager(
+        image=args.docker_image,
+        container_name=args.docker_name,
+        detect_ssh=args.detect_and_copy_host_ssh_config,
+        detect_claude=args.detect_and_copy_host_claude_config,
+    )
+
+    stream.emit(
+        EventKind.RUN_COMPLETE,
+        f"starting Docker container (image={mgr.image})",
+        exit_code=OK,
+        ok=True,
+    )
+
+    mgr.start()
+
+    # Forward the original command into the container, stripping --docker flags.
+    forwarded = ["agent-sys", "run"]
+    if args.package:
+        forwarded += ["--package", args.package]
+    if args.demo_root:
+        forwarded += ["--demo-root", args.demo_root]
+    if args.dry_run:
+        forwarded += ["--dry-run"]
+    if getattr(args, "with_broken", False):
+        forwarded += ["--with-broken"]
+    if getattr(args, "resume", False):
+        forwarded += ["--resume"]
+    if getattr(args, "allow_repo_config", False):
+        forwarded += ["--allow-repo-config"]
+    for var_item in args.var:
+        forwarded += ["--var", var_item]
+    if args.json:
+        forwarded += ["--json", args.json]
+
+    rc = mgr.exec(forwarded, workdir="/opt/Infera")
+
+    if getattr(args, "docker_rm", False):
+        mgr.stop()
+        stream.emit(
+            EventKind.RUN_COMPLETE,
+            f"container '{mgr.container_name}' removed",
+            exit_code=rc,
+            ok=rc == 0,
+        )
+
+    return rc
+
+
 def _clean(args: argparse.Namespace, stream: Stream) -> int:
     layout = layout_for(Path(args.demo_root) if args.demo_root else None)
     shutil.rmtree(layout.root / "runs", ignore_errors=True)
@@ -398,28 +525,34 @@ def _real_run(args: argparse.Namespace, stream: Stream) -> int:
     # which is how the bug was found. One call now.
     running = start_monitors(registry)
     try:
-        if args.resume:
-            resume_all(registry)
-        else:
-            _start(registry, stream)
-        _settle(registry, stream, timeout=getattr(args, "timeout", None) or _SETTLE_TIMEOUT)
+        try:
+            if args.resume:
+                resume_all(registry)
+            else:
+                _start(registry, stream)
+            _settle(registry, stream, timeout=getattr(args, "timeout", None) or _SETTLE_TIMEOUT)
+        finally:
+            # Names that did **not** come back, rather than a hang or a silent pass.
+            stragglers = running.stop(timeout=5.0)
+            if stragglers:
+                stream.emit(
+                    EventKind.RUN_COMPLETE,
+                    f"monitor loops that did not return: {sorted(stragglers)}",
+                    stragglers=sorted(stragglers),
+                    ok=False,
+                )
+        # Described AFTER the run, not before it: the subgraph does not exist
+        # until the root's main phase unfolds, so a graph printed at submit time
+        # would be one task long.
+        tasks = registry.get("task_mgr").all()
+        _emit_graph(stream, tasks, resumed=bool(args.resume))
+        _describe(registry, tasks, stream)
+        return _report(registry, stream, layout, promises)
     finally:
-        # Names that did **not** come back, rather than a hang or a silent pass.
-        stragglers = running.stop(timeout=5.0)
-        if stragglers:
-            stream.emit(
-                EventKind.RUN_COMPLETE,
-                f"monitor loops that did not return: {sorted(stragglers)}",
-                stragglers=sorted(stragglers),
-                ok=False,
-            )
-    # Described AFTER the run, not before it: the subgraph does not exist
-    # until the root's main phase unfolds, so a graph printed at submit time
-    # would be one task long.
-    tasks = registry.get("task_mgr").all()
-    _emit_graph(stream, tasks, resumed=bool(args.resume))
-    _describe(registry, tasks, stream)
-    return _report(registry, stream, layout, promises)
+        # Monitor decisions and reporting may reuse a settled executor. Once
+        # both are over, every attempt belongs to this invocation and must be
+        # disposed before Python tears down the SDK's private event loops.
+        registry.get("runner").shutdown()
 
 
 # --------------------------------------------------------------------------- #