Harden transports and state files

- Executors and relay helpers get an environment without any configured
  Plane token / webhook secret (design 8: executors hold no Plane credentials).
- Windows runner: Set-Location -LiteralPath, and a failed cd writes exit=97
  instead of running the agent in the wrong directory.
- schtasks launch: escape apostrophes and pass the script path as \"...\" so
  paths with spaces survive PowerShell 5.1 native quoting.
- State and heartbeat files are written temp + fsync + rename.
- Tests for the above plus codex run/resume command shapes and unique run dirs.

agentd/config.py

@@ -157,6 +157,17 @@ class Config:
     def token(self, terminal):
         return os.environ.get(self.token_env(terminal))
 
+    def secret_keys(self):
+        """Every env var name that carries a secret (Plane tokens, webhook secrets): never handed to executors."""
+        keys = {t["token_key"] for t in self.terminals.values()}
+        keys |= {p["webhook_secret_key"] for p in self.projects.values() if p.get("webhook_secret_key")}
+        return keys
+
+    def scrubbed_env(self):
+        """os.environ without the secrets above, for every child process agentd starts."""
+        drop = self.secret_keys()
+        return {k: v for k, v in os.environ.items() if k not in drop}
+
     def secret_env_args(self):
         """`--secret-env` arguments for agent-access.py exec, one per distinct token key."""
         keys = sorted({t["token_key"] for t in self.terminals.values()})

agentd/store.py

@@ -31,11 +31,18 @@ class Store:
         data.setdefault("paused_groups", {})
         return data
 
-    def save(self, data):
+    def _atomic_write(self, path, obj):
+        """Write temp + fsync + rename, so a reader (or a crash) never sees half a file."""
         self.dir.mkdir(parents=True, exist_ok=True)
-        tmp = self.path.with_suffix(".tmp")
-        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
-        os.replace(tmp, self.path)
+        tmp = path.with_suffix(".tmp")
+        with open(tmp, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(obj, ensure_ascii=False, indent=1))
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, path)
+
+    def save(self, data):
+        self._atomic_write(self.path, data)
 
     @contextlib.contextmanager
     def lock(self):
@@ -48,9 +55,8 @@ class Store:
             yield
 
     def heartbeat(self, info):
-        self.dir.mkdir(parents=True, exist_ok=True)
         info = dict(info, ts=time.time(), at=time.strftime("%Y-%m-%dT%H:%M:%S"))
-        (self.dir / "heartbeat.json").write_text(json.dumps(info, ensure_ascii=False, indent=1), encoding="utf-8")
+        self._atomic_write(self.dir / "heartbeat.json", info)
 
     def log(self, msg):
         self.dir.mkdir(parents=True, exist_ok=True)

agentd/transports/local.py

@@ -59,7 +59,9 @@ class LocalTransport:
     def launch(self, task, script_path, run_dir):
         # `setsid -f` double-forks: the runner leads its own session (pgid == pid, see kill) and is reparented
         # to init, so a long-running `tick --loop` never accumulates zombies.
+        # The executor must not hold Plane credentials (design §8): drop every configured secret from its env.
         subprocess.run(["setsid", "-f", "bash", script_path], cwd=run_dir, check=True, close_fds=True,
+                       env=self.cfg.scrubbed_env(),
                        stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 
     def kill(self, task, run_dir):

agentd/transports/warpgate_ssh.py

@@ -15,7 +15,8 @@ from . import TransportError
 RUNNER = r"""$ErrorActionPreference = 'Continue'
 [Console]::OutputEncoding = [Text.Encoding]::UTF8
 $OutputEncoding = [Text.Encoding]::UTF8
-Set-Location {cwd}
+try {{ Set-Location -LiteralPath {cwd} -ErrorAction Stop }} catch {{
+  Set-Content -Path '{d}\exit.txt' -Value ('exit=97 at ' + (Get-Date).ToString('s') + ' (cd failed)'); exit 97 }}
 Set-Content -Path '{d}\pid.txt' -Value $PID
 Set-Content -Path '{d}\started.txt' -Value (Get-Date).ToString('s')
 $prompt = Get-Content -Raw -Encoding UTF8 '{d}\prompt.md'
@@ -53,7 +54,7 @@ class WarpgateSshTransport:
 
     # ---------------------------------------------------------------- relay primitives
     def ps(self, script, timeout=150):
-        env = dict(os.environ, WIN10_TIMEOUT=str(timeout))
+        env = dict(self.cfg.scrubbed_env(), WIN10_TIMEOUT=str(timeout))
         try:
             out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env)
         except subprocess.TimeoutExpired:
@@ -64,7 +65,8 @@ class WarpgateSshTransport:
 
     def put(self, local, remote):
         sha = hashlib.sha256(open(local, "rb").read()).hexdigest()
-        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600)
+        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600,
+                             env=self.cfg.scrubbed_env())
         if ("remote sha256 " + sha) not in out.stdout:
             raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:]))
 
@@ -113,12 +115,18 @@ class WarpgateSshTransport:
     def runner(self, run_dir, cwd, command, stdout_name):
         return "run.ps1", RUNNER.format(cwd=self.quote(cwd), d=run_dir.replace("'", "''"), command=command, stdout=stdout_name)
 
+    def launch_script(self, task, script_path):
+        """PowerShell that (re)creates and runs the one-shot task. Values sit in '...' literals (`'` doubled). The
+        script path is wrapped in \\" so that Windows PowerShell 5.1 (legacy native quoting) hands schtasks
+        /TR `... -File "<path>"` intact even when the path contains spaces."""
+        return ("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n"
+                "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \\\"%(s)s\\\"'\n"
+                "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n"
+                "schtasks /Run /TN '%(t)s' | Out-Null\n"
+                "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task.replace("'", "''"), s=script_path.replace("'", "''")))
+
     def launch(self, task, script_path, run_dir):
-        out = self.ps("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n"
-                      "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \"%(s)s\"'\n"
-                      "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n"
-                      "schtasks /Run /TN '%(t)s' | Out-Null\n"
-                      "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task, s=script_path))
+        out = self.ps(self.launch_script(task, script_path))
         if "LAUNCHED=0" not in out:
             raise TransportError("schtasks launch of %s failed: %s" % (task, out.strip()[-300:]))