fix(worker): the Valkey pi-dispatch starts gets a per-deployment password (#468)

.env.example

@@ -59,6 +59,18 @@ PI_CONCURRENCY=3
 
 # --- Infrastructure ---
 VALKEY_URL=redis://127.0.0.1:6379
+# The Valkey's password, per deployment: `pi-dispatch init` writes a new one here (and creates this file readable by you
+# alone), and `up` / `service install` add one to a deployment that has none and restart its Valkey with it (the queue is
+# kept). Every client sends it: the worker, the receiver, the CLI, the panel and doctor. Without it any local account can
+# read, enqueue or delete this deployment's jobs. Hex or base64url only (A-Z a-z 0-9 - _), at least 16 characters.
+# A key of its own, never inside VALKEY_URL, which doctor prints and the panel's pointer file stores; a VALKEY_URL with
+# its own password (a managed Valkey) is left as it is, and that password wins. With PI_VALKEY_SHARED=1 below, set it to
+# the password of the account whose Valkey it is. docker compose reads it with --env-file .env (docs/podman.md, README).
+# VALKEY_PASSWORD=
+# docker compose only: the port deploy/docker-compose.yml publishes its Valkey on, 127.0.0.1:<this>. Default 6379. Keep it
+# equal to VALKEY_URL's port: `pi-dispatch up` and /dispatch setup write it here when that is not 6379 (never over a value
+# already set), and doctor flags one that disagrees. Not read by the worker, the receiver or the CLI, which dial VALKEY_URL.
+# PI_VALKEY_PORT=
 # podman venue: the worker (at every start), `service install`, `up` and doctor take a Valkey that answers VALKEY_URL only when THIS account
 # (or one of its containers) holds it. One held by another account, by root (docker-proxy, a rootful container) or by a
 # system service is refused, naming its owner: give this account its own port instead (VALKEY_URL=redis://127.0.0.1:6380).
@@ -269,7 +281,7 @@ PI_SCHEDULER_STALL_MAX=2
 # --- Egress policy: what a job container may reach on the network (docs/egress.md) ---
 # ON by default. Every job runs on its own --internal Docker network whose only other member is an allowlist
 # proxy, with no route off this host, and a job whose policy cannot serve it is refused BEFORE it spends a slot.
-# START THE PROXY:  docker compose -f deploy/docker-compose.yml --profile egress up -d
+# START THE PROXY:  docker compose --env-file .env -f deploy/docker-compose.yml --profile egress up -d
 # (on the rootless podman backend, `pi-dispatch up` or `pi-dispatch service install` starts it as a Quadlet unit instead)
 # Until you do, every job is refused pre-spend (loud, free, and naming that command). The hosts live in
 # egress-allowlist.conf next to this file (`pi-dispatch init` writes it, and never overwrites it).

README.md

@@ -93,7 +93,9 @@ The same setup as plain commands. No clone needed:
 mkdir my-dispatch && cd my-dispatch
 npx @edgehero/pi-dispatch up   # one consented pass: pulls the job image, starts Valkey,
                                #   scaffolds the config files, fills in the .env lines that have
-                               #   no value yet, runs the doctor preflight.
+                               #   no value yet (VALKEY_PASSWORD among them: the queue's own
+                               #   password, so no other account on the host can read it),
+                               #   runs the doctor preflight.
                                #   Every docker action shows its command and asks first; the .env
                                #   lines are filled without a prompt, and never overwritten.
 #  edit .env and set your provider key. Already logged into pi with an API key? Leave it blank.
@@ -669,10 +671,15 @@ which is what it offers right after the credentials step):
 1. **Webhook receiver on the host** (lowest latency): `npx pi-dispatch-receiver` from your deployment
    folder, or `pi-dispatch service install --receiver` to run it as a user-level service. Your reverse
    proxy or tunnel does the public exposure.
-2. **Webhook receiver in a container**: `docker compose -f deploy/docker-compose.yml --profile receiver
+2. **Webhook receiver in a container**: `docker compose --env-file .env -f deploy/docker-compose.yml --profile receiver
    up -d` runs the prebuilt
    [`ghcr.io/edgehero/pi-dispatch-receiver`](https://github.com/edgehero/pi-dispatch/pkgs/container/pi-dispatch-receiver)
-   beside Valkey, triggers mounted read-only, no docker socket anywhere.
+   beside Valkey, triggers mounted read-only, no docker socket anywhere. In a folder `/dispatch setup` built, the
+   compose file sits in `deploy/` of that folder, and the command it runs names the folder's own project, `-p <folder
+   name>` (lower case, only letters, digits, `_` and `-`), and, where `pi-dispatch up` had already started the
+   deployment's Valkey, `-f deploy/docker-compose.valkey.yml`, which gives compose's Valkey that same volume so the
+   worker and the receiver share one queue. Run it the same way afterwards. Setup hands over only a
+   `pi-dispatch-valkey` that is this folder's, and refuses while another deployment's Valkey uses that volume.
 3. **No public URL at all**: `pi-dispatch-receiver poll` fetches issue events, comments and PRs over TLS
    with your own credential (conditional requests, nearly free against the rate limit). Same gates, same
    queue, about 60 seconds of latency, zero public surface. Pair with `setup github --no-webhook`. A

SECURITY.md

@@ -137,7 +137,7 @@ Jobs are a **trigger × target** matrix, and the triggers do not share a threat
   prefer (its own bullet below states the consequences). One implementation detail, recorded because the code
   records it: Azure's constant-time compare returns early on a length mismatch, so the secret's **length**
   leaks even though its content does not.
-  The receiver may run **containerised** (`docker compose -f deploy/docker-compose.yml --profile receiver up`,
+  The receiver may run **containerised** (`docker compose --env-file .env -f deploy/docker-compose.yml --profile receiver up`,
   issue #82; the `-f` is load-bearing, because that file's relative paths resolve against `deploy/` and not
   your cwd). That changes none of the
   verification above: it is the same code, the container mounts `triggers.json` read-only and **no docker

admin/src/deployment-pointer.mjs

@@ -181,6 +181,9 @@ function layerPointer(env, fs) {
     ownedKeys.add(key);
     applied.push(key);
   }
+  // Issue #468: the panel's Valkey clients take VALKEY_PASSWORD (and PI_VALKEY_SHARED) from the POINTED-AT deployment's
+  // `.env` (`panelValkeyContext`, read-model.mjs, from `pointerState()` below, through issue #471's one reader). The
+  // pointer still carries no credential, and the password is never copied into the environment.
   return { applied };
 }
 

admin/src/index.ts

@@ -74,6 +74,8 @@ import {
   cancelHeldJob,
   listRunIds,
   setQueuePaused,
+  killSwitchSet,
+  installPanelValkeyContext,
   writeSettings,
   writeTriggers,
   writePauseWindows,
@@ -256,6 +258,9 @@ export default function admin(pi: ExtensionAPI): void {
     // Deliberately swallowed: applyDeploymentPointer degrades internally ({ ignored } + notice), so
     // anything reaching here is unexpected -- and still must not take the whole extension down.
   }
+  // Issue #468 over issue #471: every Valkey client of the panel takes its password (and its facts about the
+  // deployment's Valkey) from the pointer's folder through the one `.env` reader the panel has, never another.
+  installPanelValkeyContext();
 
   pi.registerCommand("dispatch", {
     // The operator-visible summary; a subcommand once went missing from it while USAGE carried it,
@@ -384,17 +389,20 @@ function registerTools(pi: ExtensionAPI): void {
     executionMode: "sequential",
     parameters: Type.Object({}),
     async execute() {
-      const paths = resolvePaths(deploymentEnv());
-      const res = await setQueuePaused({ url: paths.valkeyUrl, paused: true });
-      if (res.unreachable) {
-        throw new Error(`could not reach the queue at ${paths.valkeyUrl}: ${res.unreachable}`);
-      }
-      // A PARTIAL result is the dangerous one: some queues switched, one did not, and reporting a bare
-      // "paused" for it tells an operator the deployment is stopped while a host keeps spending.
-      if (res.partial) {
-        throw new Error(`paused ${res.partial.done.join(", ")} but FAILED at ${res.partial.failed} — the deployment is half paused: ${res.partial.error}`);
+      // The CLI's kill-switch rule (PR #475's review, round 3): both Valkeys when this process's VALKEY_URL and the
+      // deployment .env's disagree, every URL shown without its userinfo.
+      const set = await killSwitchSet({ paused: true });
+      if (set.refused) throw new Error(set.refused);
+      const lines = [];
+      for (const { shown, res } of set.results) {
+        const at = set.results.length > 1 ? ` at ${shown}` : "";
+        if (res.unreachable && !res.partial) throw new Error(`could not reach the queue at ${shown}: ${res.unreachable}`);
+        // A PARTIAL result is the dangerous one: some queues switched, one did not, and reporting a bare
+        // "paused" for it tells an operator the deployment is stopped while a host keeps spending.
+        if (res.partial) throw new Error(`paused ${res.partial.done.join(", ")}${at} but FAILED at ${res.partial.failed}: the deployment is half paused: ${res.partial.error}`);
+        lines.push(`paused${at}${res.queues.length > 1 ? ` (${res.queues.length} queues)` : ""}`);
       }
-      return toolText(`paused${res.queues.length > 1 ? ` (${res.queues.length} queues)` : ""}`);
+      return toolText(`${set.disagreement ? `${set.disagreement}: paused both. ` : ""}${lines.join("; ")}`);
     },
   });
 
@@ -405,10 +413,11 @@ function registerTools(pi: ExtensionAPI): void {
     executionMode: "sequential",
     parameters: Type.Object({}),
     async execute() {
-      const paths = resolvePaths(deploymentEnv());
-      const res = await setQueuePaused({ url: paths.valkeyUrl, paused: false });
-      if (res.unreachable) {
-        throw new Error(`could not reach the queue at ${paths.valkeyUrl}: ${res.unreachable}`);
+      const set = await killSwitchSet({ paused: false });
+      if (set.refused) throw new Error(set.refused);
+      const { shown, res } = set.results[0];
+      if (res.unreachable && !res.partial) {
+        throw new Error(`could not reach the queue at ${shown}: ${res.unreachable}`);
       }
       if (res.partial) {
         throw new Error(`resumed ${res.partial.done.join(", ")} but FAILED at ${res.partial.failed} — the deployment is half resumed: ${res.partial.error}`);
@@ -1372,26 +1381,36 @@ async function dispatch(pi: ExtensionAPI, args: string, rawCtx: any): Promise<vo
     case "pause":
     case "resume": {
       const paused = sub === "pause";
-      const res = await setQueuePaused({ url: paths.valkeyUrl, paused });
-      if (res.unreachable) {
-        // A HALF switched deployment gets its own message. "Could not reach Valkey" reads as "nothing
-        // happened", and walking away from a fleet with one host stopped and another spending is the one
-        // outcome this switch exists to prevent.
+      // The CLI's kill-switch rule (PR #475's review, round 3): a pause pauses BOTH Valkeys when this process's
+      // VALKEY_URL and the deployment .env's disagree, a resume refuses then, and every URL is shown without userinfo.
+      const set = await killSwitchSet({ paused });
+      if (set.refused) {
+        notify?.(set.refused, "error");
+        return;
+      }
+      if (set.disagreement) notify?.(`${set.disagreement}: pausing both`, "warning");
+      for (const { shown, res } of set.results) {
+        const at = set.results.length > 1 ? ` at ${shown}` : "";
+        if (res.unreachable) {
+          // A HALF switched deployment gets its own message. "Could not reach Valkey" reads as "nothing
+          // happened", and walking away from a fleet with one host stopped and another spending is the one
+          // outcome this switch exists to prevent.
+          notify?.(
+            res.partial
+              ? `${paused ? "paused" : "resumed"} ${res.partial.done.join(", ")}${at} but FAILED at ${res.partial.failed}: the deployment is half ${paused ? "paused" : "resumed"}`
+              : `could not reach Valkey at ${shown}: is it running? (docker compose up)`,
+            "error",
+          );
+          continue;
+        }
+        const span = res.blind ? ` (fleet unknown: ${res.blind}, only ${res.queues[0]})` : res.queues.length > 1 ? ` (${res.queues.length} queues)` : "";
         notify?.(
-          res.partial
-            ? `${paused ? "paused" : "resumed"} ${res.partial.done.join(", ")} but FAILED at ${res.partial.failed} — the deployment is half ${paused ? "paused" : "resumed"}`
-            : `could not reach Valkey at ${paths.valkeyUrl} — is it running? (docker compose up)`,
-          "error",
+          paused
+            ? `paused${at}: worker will stop taking new jobs (jobs still enqueue; durable, survives restart)${span}`
+            : `resumed${at}${span}`,
+          "info",
         );
-        return;
       }
-      const span = res.blind ? ` (fleet unknown: ${res.blind}, only ${res.queues[0]})` : res.queues.length > 1 ? ` (${res.queues.length} queues)` : "";
-      notify?.(
-        paused
-          ? `paused — worker will stop taking new jobs (jobs still enqueue; durable, survives restart)${span}`
-          : `resumed${span}`,
-        "info",
-      );
       return;
     }
     case "set": {

admin/src/read-model.mjs

@@ -32,7 +32,8 @@ import { HELD_SET, jobKey } from "@edgehero/pi-dispatch/wait-state";
 // The subscriptions validator is shared for the same anti-drift reason: the admin prices finished runs
 // against the exact schema the file declares, and re-deriving it here is how the two would disagree.
 import { parseSubscriptions, SUBSCRIPTIONS_VERSION } from "@edgehero/pi-dispatch/subscriptions";
-import { parseConnection, makeRedisClient } from "@edgehero/pi-dispatch/connection";
+import { parseConnection, makeRedisClient, killSwitchValkeyUrls, urlShown, useValkeyContext, valkeyContextFromKeys, VALKEY_CONTEXT_KEYS } from "@edgehero/pi-dispatch/connection";
+import { pointerState } from "./deployment-pointer.mjs";
 import { readLiveHosts } from "@edgehero/pi-dispatch/host-registry";
 import { QUEUE, makeQueue, enqueueLocalJob, fleetQueueNames, hostQueueName, discoverHostQueues, unionQueueNames } from "@edgehero/pi-dispatch/queue";
 import { hostsIn, mergeRuns, readMirroredRuns } from "@edgehero/pi-dispatch/run-mirror";
@@ -312,6 +313,68 @@ export async function readFleetQueues({ url, redisFn = makeRedisClient, timeoutM
   }
 }
 
+/**
+ * Issue #468 with issue #471's rule: where the panel's Valkey clients stand (`valkeyClientContext`'s shape), from the
+ * pointer's deployment folder ONLY, read through the one reader the panel has (`deploymentServiceEnv`, one descriptor,
+ * nothing from a file another account can change): its VALKEY_PASSWORD, PI_VALKEY_SHARED and PI_BACKENDS, and its
+ * VALKEY_URL beside pi's own. pi's own environment is taken without what the pointer layered into it (the pointer's
+ * VALKEY_URL is the wizard's snapshot of the file, which the file outranks), so a disagreement is between the operator's
+ * export and the file. No pointer: pi's environment alone, and no file. A file that could not be taken from is the
+ * context's `error`, as `valkeyClientContext` says one it could not read.
+ */
+export function panelValkeyContext({ env = process.env, pointerDir = null, owned = [], fs = nodeFs, platform = process.platform, uid = process.geteuid?.() } = {}) {
+  const own = {};
+  for (const [key, value] of Object.entries(env)) if (!owned.includes(key)) own[key] = value;
+  if (pointerDir === null) return valkeyContextFromKeys({ env: own, platform });
+  // The file's own values (resolved against an empty shell, so `fromFile` is exactly what the file says).
+  const res = deploymentServiceEnv({ env: {}, dir: pointerDir, keys: VALKEY_CONTEXT_KEYS, platform, fs, uid });
+  const error = res.untrusted
+    ? `${res.path} ${res.untrusted}`
+    : res.unreadable
+      ? `${res.path} could not be read (${res.unreadable})`
+      : res.hazardSkipped.length > 0
+        ? `${res.path} line ${res.hazard.line} ${res.hazard.what}`
+        : res.unread.length > 0
+          ? `${res.path} assigns ${res.unread.map((u) => `${u.key} (line ${u.line})`).join(", ")} in a form the service's loader may read differently`
+          : null;
+  return valkeyContextFromKeys({ env: own, envPath: res.path, fileKeys: res.fromFile, error, platform });
+}
+
+/**
+ * Make `panelValkeyContext`, from the pointer's current state, the context every Valkey client of this process gets
+ * when its caller names none (`useValkeyContext`), so no client of the panel reads a `.env` by any other way. Read per
+ * client, like `panelEnv`, so an edit to the file reaches the next command.
+ */
+export function installPanelValkeyContext({ env = process.env } = {}) {
+  useValkeyContext(() => {
+    const { deploymentDir, owned } = pointerState();
+    return panelValkeyContext({ env, pointerDir: deploymentDir, owned });
+  });
+}
+
+/**
+ * The panel's kill switch (PR #475's review, round 3): the CLI's rule, from the one resolver both import
+ * (`killSwitchValkeyUrls`), over `panelValkeyContext` (issue #471: the pointer's folder only, through the one reader).
+ * The panel used this process's VALKEY_URL alone, so a stale one paused a Valkey the worker does not drain and said
+ * "paused". When pi's own VALKEY_URL and the deployment .env's disagree, a PAUSE pauses both, and a RESUME, which would
+ * start spending, is refused until they agree. Returns `{ refused }`, or `{ results: [{ url, shown, res }],
+ * disagreement }` with each URL as `urlShown` prints it, never with its userinfo; every client sends the deployment's
+ * password by that same context.
+ */
+export async function killSwitchSet({ paused, env = process.env, setFn = setQueuePaused, resolveFn = killSwitchValkeyUrls, stateFn = pointerState, fs = nodeFs, uid = process.geteuid?.() } = {}) {
+  const { deploymentDir, owned } = stateFn();
+  const context = panelValkeyContext({ env, pointerDir: deploymentDir, owned, fs, uid });
+  const picked = resolveFn({ env, context });
+  if (picked.error) return { refused: picked.error };
+  if (picked.urls.length > 1 && !paused) {
+    return { refused: `${picked.disagreement}: resume would start jobs on one of them, so it resumes neither. Make them agree (the .env is what the service reads), or run: pi-dispatch resume --valkey-url <url>` };
+  }
+  const conn = { parseConnectionFn: (u, o = {}) => parseConnection(u, { ...o, context }), redisFn: (u, o = {}) => makeRedisClient(u, { ...o, context }) };
+  const results = [];
+  for (const url of picked.urls) results.push({ url, shown: urlShown(url), res: await setFn({ url, paused, ...conn }) });
+  return { results, disagreement: picked.urls.length > 1 ? picked.disagreement : null };
+}
+
 export async function setQueuePaused({ url, paused, makeQueueFn = makeQueue, parseConnectionFn = parseConnection, redisFn = makeRedisClient, timeoutMs = 2500 } = {}) {
   const opened = [];
   try {