#!/usr/bin/env python3 """dsh_worker - shell out one task to a dsh (DeepSeek Harness) headless agent and hand the reply back, or an honest, named REFUSAL if it did not get one. python tools/dsh_worker.py "task text on the command line" python tools/dsh_worker.py --file task.md echo "task text" | python tools/dsh_worker.py - python tools/dsh_worker.py --plain "task text" # reply/reason only, no JSON Prints one JSON object to stdout and sets the exit code to match: {"ok": true, "reply": ""} -> exit 0 {"ok": false, "refused": true, "reason": ""} -> exit 1 `--plain` is the only non-default output shape: it prints just the reply text (on success) or just the reason text (on refusal), for a human running this at a terminal. Every caller that is code, not a person, should use the default JSON shape -- that is the "structured mode" a script can rely on to tell a reply from a diagnostic without parsing prose. `--json` also exists, as a no-op, for a caller that wants to say so explicitly. WHY THIS EXISTS ---------------- The owner's instruction (2026-09-02): stand up dsh and let it take the routine work a Claude subagent would otherwise do, because dsh's models are cheaper. `dsh --profile headless ` is dsh's own one-shot mode -- "Answer one task, print the final assistant message, and exit" (its own --help text) -- which is exactly the CLI-worker shape a coordinator can shell out to. There is no dsh SDK or importable client on this machine; `tools/aiextract_run.py` talks to dsh's session.* RPC directly because it needs to keep a session open and poll it (see that file's module docstring for the wire contract), but a one-shot worker has no reason to reimplement that protocol when the CLI already does the same round trip in one call. THE RULE THIS FILE IS BUILT AROUND (CLAUDE.md SS3 / SS5): "could not judge" and "judged, found nothing" are different answers, and this file must never collapse them. A dsh call that errors, times out, or returns a blank reply is reported as a REFUSAL naming what went wrong -- never as an empty-but-"ok" result, which is indistinguishable from "the task legitimately produced nothing" and is exactly the silent-pass failure mode CLAUDE.md's skip-probe table exists to prevent for Ticket QC checks. Same discipline, different place it can bite. THE PROFILE IS CONFIGURABLE, NOT HARDCODED. The owner's actual working setup (confirmed live 2026-09-02) does not use dsh's own `deepseek-official` route -- it routes through an "opencode go" API instead: dsh's `headless` profile plus `~/.dsh/settings.yaml`'s `agent-default-model` override, which points the `llm-pi-ai` plugin's `openocde-go` provider (yes, that's the literal id in his config -- do not "fix" the letter order, it is not this file's config to rename) at `deepseek-v4-flash`, with the API key held by dsh's own credentials service (`~/.dsh/.credentials.yaml`, keyed `OPENOCDE_GO_API_KEY` -- never an OS environment variable on this machine). None of that is this file's business to know beyond "which profile to boot": the profile is `DEFAULT_PROFILE`, overridable with `DSH_WORKER_PROFILE` or `--profile`, and whichever route that profile's own settings.yaml points at is what runs. Someone else's machine with a real DEEPSEEK_API_KEY and no settings.yaml override gets dsh's stock `deepseek-official` route through the exact same `headless` profile, with no code change here. THE WINDOWS ARGV TRAP: dsh installs on Windows as a `dsh.CMD` shim that launches `node \\node_modules\\@deepseek-ai\\dsh\\lib\\bin.js`. Handing that shim a task argument through Python's subprocess module (a plain list, no shell=True) still means Windows routes it through cmd.exe's own command-line parser to resolve the .CMD file, and cmd.exe treats a newline inside an argument as a command separator -- **silently truncating the task at the first newline, exit 0, no error**. Measured 2026-09-02: a two-line task (`"line one\\nline two with quotes"`) reached dsh as `"line one"` alone, and dsh answered a real, well-formed reply to that truncated text -- which is the most dangerous shape of this bug, because nothing about the reply looks wrong on its own. Any task longer than one line (which is most real work) hits this. So on Windows, when `which()` resolves `dsh` to a `.cmd`/`.bat` shim, `_resolve_launch_cmd` skips the shim and invokes `node.exe` on the underlying `bin.js` directly -- the same argv element, no cmd.exe in the middle, newlines and all. Off Windows, or when the shim can't be resolved to its underlying script, the shim is invoked as before. Standard library only. CPython 3.8+ -- this runs on the host, not inside Revit; see CLAUDE.md's "Write for both Pythons" -- this file is exempt from the IronPython 2.7 rules because nothing under WillTools.extension/ imports it. """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import time DEFAULT_PROFILE = os.environ.get('DSH_WORKER_PROFILE', 'headless') DEFAULT_TIMEOUT_S = int(os.environ.get('DSH_WORKER_TIMEOUT_S', '300')) DSH_EXE = 'dsh' class DshRefusal(Exception): """The task did not get a usable reply. `reason` is what a caller prints.""" def _find_dsh(which=shutil.which): path = which(DSH_EXE) if not path: raise DshRefusal( 'dsh is not on PATH -- install it with ' '"npm install -g @deepseek-ai/dsh" and see docs/dsh-worker.md') return path def _resolve_launch_cmd(dsh_path): """Return the argv prefix to actually invoke `dsh` with. On Windows, `which()` resolves `dsh` to a `.CMD`/`.BAT` shim that forwards its arguments through cmd.exe to `node \\ node_modules\\@deepseek-ai\\dsh\\lib\\bin.js`. cmd.exe's own command-line parsing truncates any argument at its first newline -- see the module docstring's "Windows argv trap" -- so a multi-line task silently loses everything after line one. Skip the shim: call node directly on the resolved bin.js, so the task text reaches dsh exactly as Python handed it to subprocess, newlines included. Anywhere this can't be resolved (not Windows, not a .cmd/.bat shim, or the expected files aren't where the shim would put them) this returns `[dsh_path]` unchanged -- the previous, always-correct-off-Windows behaviour. """ if os.name != 'nt': return [dsh_path] if not dsh_path.lower().endswith(('.cmd', '.bat')): return [dsh_path] shim_dir = os.path.dirname(dsh_path) bin_js = os.path.join(shim_dir, 'node_modules', '@deepseek-ai', 'dsh', 'lib', 'bin.js') if not os.path.isfile(bin_js): return [dsh_path] node_exe = os.path.join(shim_dir, 'node.exe') if not os.path.isfile(node_exe): node_exe = shutil.which('node') if not node_exe: return [dsh_path] return [node_exe, bin_js] def run_task(task, profile=DEFAULT_PROFILE, timeout_s=DEFAULT_TIMEOUT_S, which=shutil.which, runner=subprocess.run, clock=time.time): """Run one task through a dsh headless profile. Returns the reply text. Raises DshRefusal, always with a `reason` that names the layer that failed, on any of: dsh missing, a non-zero exit, a timeout, or a zero exit with an empty/whitespace-only reply -- the last of those being the one an unwary caller would mistake for success. `which` and `runner` are injected the same way `tools/aiextract_run.py` injects `transport` -- so a test can fake the subprocess boundary without a real dsh install or a real API key. """ task = (task or '').strip() if not task: raise DshRefusal('empty task text -- nothing to send to dsh') dsh_path = _find_dsh(which) launch_cmd = _resolve_launch_cmd(dsh_path) cmd = launch_cmd + ['--profile', profile, task] started = clock() try: proc = runner(cmd, capture_output=True, text=True, timeout=timeout_s, encoding='utf-8', errors='replace') except subprocess.TimeoutExpired: raise DshRefusal( 'dsh --profile {0} did not answer within {1}s' .format(profile, timeout_s)) except OSError as exc: raise DshRefusal('could not start dsh ({0}): {1}' .format(dsh_path, exc)) elapsed = clock() - started if proc.returncode != 0: stderr = (proc.stderr or '').strip() raise DshRefusal( 'dsh --profile {0} exited {1} after {2:.1f}s: {3}' .format(profile, proc.returncode, elapsed, stderr or '(no stderr)')) reply = (proc.stdout or '').strip() if not reply: # THE TRAP: exit 0 with nothing on stdout is NOT success. Treat it # exactly like the CLAUDE.md skip-probe table treats an empty check # result with no probe registered -- as a defect to surface, never # as a quiet "ran, found nothing" pass. See module docstring. stderr = (proc.stderr or '').strip() raise DshRefusal( 'dsh --profile {0} exited 0 but printed no reply after {1:.1f}s ' '(stderr: {2}) -- could not judge, not judged-empty' .format(profile, elapsed, stderr or '(none)')) return reply def main(argv=None): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument('task', nargs='?', help='task text, or "-" to read it from stdin') ap.add_argument('--file', help='read the task text from this file') ap.add_argument('--profile', default=DEFAULT_PROFILE, help='dsh profile to boot (default: {0}; also settable ' 'with $DSH_WORKER_PROFILE)'.format(DEFAULT_PROFILE)) ap.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT_S, help='seconds to wait before refusing (default: {0}; ' 'also settable with $DSH_WORKER_TIMEOUT_S)' .format(DEFAULT_TIMEOUT_S)) ap.add_argument('--json', action='store_true', help='structured JSON output (default; this flag is a ' 'no-op kept for callers that want to say so ' 'explicitly)') ap.add_argument('--plain', action='store_true', help='print only the reply text (success) or reason ' 'text (refusal), no JSON -- for a human at a ' 'terminal, not for a caller that is code') args = ap.parse_args(argv) if args.file: with open(args.file, 'r', encoding='utf-8') as fh: task = fh.read() elif args.task == '-' or (args.task is None and not sys.stdin.isatty()): task = sys.stdin.read() elif args.task is not None: task = args.task else: reason = ('no task given -- pass it as an argument, --file , ' 'or on stdin') if args.plain: print(reason) else: print(json.dumps({'ok': False, 'refused': True, 'reason': reason})) return 1 try: reply = run_task(task, profile=args.profile, timeout_s=args.timeout) except DshRefusal as exc: if args.plain: print(str(exc)) else: print(json.dumps({'ok': False, 'refused': True, 'reason': str(exc)})) return 1 if args.plain: print(reply) else: print(json.dumps({'ok': True, 'reply': reply})) return 0 if __name__ == '__main__': raise SystemExit(main())