fix: guard remote apply-patch context

docs/reference/cli.md

@@ -132,9 +132,11 @@ core 只允许声明了 `host.ssh` capability 的 provider 使用 `ssh` 透传
 
 ssh-like 远端命令如果出现 `kex_exchange_identification`、`Connection closed by remote host`、provider session timeout 或 exit code 255，CLI 会在原始 stderr 后追加一行 `UNIDESK_SSH_HINT { ... }`。该 JSON 不回显原始远端命令，只包含 `code=ssh-like-command-friction`、`trigger`、`try` 和 `triage`；`try` 固定指向 stdin script 形态，避免把一次 ssh-like 解析/握手摩擦误读成 D601 SSH 整体不可用。
 
-`ssh <providerId>` 会在远端会话启动时注入 `/tmp/unidesk-ssh-tools/apply_patch`、`/tmp/unidesk-ssh-tools/glob` 和 `/tmp/unidesk-ssh-tools/skill-discover`，并把该目录加入远端 `PATH`。`apply_patch` 接受标准 `*** Begin Patch` / `*** End Patch` patch 格式，便于通过 SSH 透传编辑远端仓库文件；`glob` 在远端用 Python 执行路径匹配，避免依赖 shell glob 展开；`skill-discover` 用于列出远端 Linux/WSL 与 Windows skill。目标节点需要具备 `python3` 和 `base64`。注入工具只写 `/tmp/unidesk-ssh-tools`，不修改目标仓库，交互式 shell 和远端命令都可以直接调用这些工具。
+`ssh <providerId>` 会在远端会话启动时注入 `/tmp/unidesk-ssh-tools/apply_patch`、`/tmp/unidesk-ssh-tools/glob` 和 `/tmp/unidesk-ssh-tools/skill-discover`，并把该目录加入远端 `PATH`。`apply_patch` 是统一的 sh helper，接受标准 `*** Begin Patch` / `*** End Patch` patch 格式，便于通过 SSH 透传编辑远端仓库文件；`glob` 在远端用 Python 执行路径匹配，避免依赖 shell glob 展开；`skill-discover` 用于列出远端 Linux/WSL 与 Windows skill。目标节点需要具备 `sh`、`base64`，并在使用 `glob`/`skill-discover` 时具备 `python3`。注入工具只写 `/tmp/unidesk-ssh-tools`，不修改目标仓库，交互式 shell 和远端命令都可以直接调用这些工具。
 
-如果只是远端打小补丁，不需要再手写 `ssh D601 'apply_patch' < patch.diff` 这种命令拼接；正式入口是 `bun scripts/cli.ts ssh D601 apply-patch < patch.diff`。`apply-patch` 与 `patch` 等价，附加参数会原样透传给远端 `apply_patch`，例如 `bun scripts/cli.ts ssh D601 apply-patch --help`。标准单命令用法如下，不需要先创建本地 patch 临时文件：
+`apply_patch` 默认拒绝低上下文 update hunk：空搜索/纯插入无锚点、只在插入点前有上下文而没有插入点后上下文、或同一 hunk search 在目标文件中匹配多个位置时，都会结构化失败并提示补充上下文。成功应用时每个 hunk 会在 stderr 输出 `apply_patch: hunk N matched path:line`，用于复核实际落点；只有人工确认确实需要文件开头插入、重复上下文或其他模糊改写时，才允许给 `apply-patch --allow-loose`。
+
+如果只是远端打小补丁，不需要再手写 `ssh D601 'apply_patch' < patch.diff` 这种命令拼接；正式入口是 `bun scripts/cli.ts ssh D601 apply-patch < patch.diff`。`apply-patch` 与 `patch` 等价，附加参数会原样透传给远端 `apply_patch`，例如 `bun scripts/cli.ts ssh D601 apply-patch --help` 或 `bun scripts/cli.ts ssh D601 apply-patch --allow-loose < reviewed.patch`。标准单命令用法如下，不需要先创建本地 patch 临时文件：
 
 ```bash
 bun scripts/cli.ts ssh D601 apply-patch <<'PATCH'
@@ -217,7 +219,7 @@ bun scripts/cli.ts ssh D601:k3s:hwlab-dev:hwlab-cloud-api apply-patch <<'PATCH'
 PATCH
 ```
 
-`logs` operation 默认是有界读取；`--follow`/`-f` 会被拒绝，防止 CLI 长时间占用维护桥。目标 route 后面直接跟普通命令时，CLI 会把 argv 放到 `kubectl exec --` 后；显式 `exec` operation 可用于让命令边界更清晰。如果命令本身需要复杂 shell 语法，优先改用 `script` operation，把脚本走 stdin，而不是把 `kubectl exec ... -- sh -c ...` 放进远端命令字符串。pod 内 `apply-patch` operation 临时注入的是 `sh` 版 helper，不要求目标容器自带 `python3`、`node` 或仓库里的工具脚本；它面向文本热修复，不用于大文件或二进制改写。
+`logs` operation 默认是有界读取；`--follow`/`-f` 会被拒绝，防止 CLI 长时间占用维护桥。目标 route 后面直接跟普通命令时，CLI 会把 argv 放到 `kubectl exec --` 后；显式 `exec` operation 可用于让命令边界更清晰。如果命令本身需要复杂 shell 语法，优先改用 `script` operation，把脚本走 stdin，而不是把 `kubectl exec ... -- sh -c ...` 放进远端命令字符串。pod 内 `apply-patch` operation 使用同一个 sh helper，不要求目标容器自带 `python3`、`node` 或仓库里的工具脚本；它面向文本热修复，不用于大文件或二进制改写。
 
 `ssh <providerId> argv <command> [args...]` 是通用 argv 安全拼接入口；`exec` 是同义入口。它是非交互远端单进程命令的默认成功路径，不需要 shell 管道时直接传命令和参数，例如 `bun scripts/cli.ts ssh D601 argv true`。需要管道、重定向、变量展开或多条命令时，优先改用 `ssh <providerId> script <<'SCRIPT'`。`find`、`glob` 和 `apply-patch` 有专用入口；`rg`、`grep`、`sed`、`nl`、`stat`、`du`、`ls`、`cat`、`head`、`tail`、`wc` 和 `pwd` 可以直接作为 `ssh` 子命令使用，CLI 会对每个 argv token 做 shell quoting。旧的自由 ssh-like 远端命令入口只保留为近似原生 ssh 的人工兼容路径。
 

scripts/src/help.ts

@@ -148,7 +148,7 @@ export function sshHelp(): unknown {
     usage: [
       "bun scripts/cli.ts ssh <route>",
       "bun scripts/cli.ts ssh <providerId> argv <command> [args...]",
-      "bun scripts/cli.ts ssh <providerId> apply-patch < patch.diff",
+      "bun scripts/cli.ts ssh <providerId> apply-patch [--allow-loose] < patch.diff",
       "bun scripts/cli.ts ssh <providerId> py [script-args...] < script.py",
       "bun scripts/cli.ts ssh <providerId> script [--shell sh|bash] [script-args...] <<'SCRIPT'",
       "bun scripts/cli.ts ssh <providerId> skills [--scope all|wsl|windows] [--limit N]",
@@ -169,6 +169,7 @@ export function sshHelp(): unknown {
     notes: [
       "ssh --help and ssh <route> --help print this JSON help and never open an interactive session.",
       "For non-interactive remote commands, prefer argv for a single process and script/stdin for shell logic.",
+      "apply-patch rejects low-context update hunks by default, reports the matched file:line for each hunk on stderr, and only accepts --allow-loose when the caller has manually reviewed an intentionally ambiguous insertion.",
       "script defaults to target /bin/sh and inherits provider proxy variables such as HTTP_PROXY/HTTPS_PROXY/ALL_PROXY/NO_PROXY; use --shell bash only for bash syntax such as pipefail, arrays, or [[ ... ]], not as a proxy workaround.",
       "Route syntax is `{provider}:{plane}[:{scope...}] {operation} [operation-args...]`: the first argv token locates a distributed target only, and every following token belongs to the operation parser. Host workspace routes use `<provider>:/absolute/workspace`; native k3s providers such as D601 and G14 use <provider>:k3s for the control plane, <provider>:k3s:<namespace>:<workload> for a workload, and <provider>:k3s:<namespace>:<workload>/<pod-workspace> for a pod workspace.",
       "Do not put operation names in any colon route segment, including nested k3s namespace/workload/container segments.",

scripts/src/ssh.ts

@@ -62,152 +62,11 @@ const legacyK3sOperationRouteSegments = new Set([
   "api-versions",
 ]);
 
-const remoteApplyPatchSource = String.raw`#!/usr/bin/env python3
-import sys
-from pathlib import Path
-
-
-def die(message):
-    print(f"apply_patch: {message}", file=sys.stderr)
-    sys.exit(1)
-
-
-def write_file(path, text):
-    target = Path(path)
-    target.parent.mkdir(parents=True, exist_ok=True)
-    target.write_text(text)
-
-
-def move_file(source, destination):
-    source_path = Path(source)
-    destination_path = Path(destination)
-    if not source_path.exists():
-        die(f"file not found: {source}")
-    if destination_path.exists():
-        die(f"target file already exists: {destination}")
-    destination_path.parent.mkdir(parents=True, exist_ok=True)
-    source_path.rename(destination_path)
-
-
-def read_file(path):
-    try:
-        return Path(path).read_text()
-    except FileNotFoundError:
-        die(f"file not found: {path}")
-
-
-def apply_update(path, body):
-    old = read_file(path)
-    output = []
-    search_from = 0
-    index = 0
-    while index < len(body):
-        if body[index].startswith("*** End of File"):
-            index += 1
-            continue
-        if not body[index].startswith("@@"):
-            die(f"expected hunk header in {path}")
-        index += 1
-        sequence = []
-        while index < len(body) and not body[index].startswith("@@"):
-            line = body[index]
-            index += 1
-            if line.startswith("*** End of File"):
-                continue
-            if line.startswith(" "):
-                sequence.append((" ", line[1:]))
-            elif line.startswith("-"):
-                sequence.append(("-", line[1:]))
-            elif line.startswith("+"):
-                sequence.append(("+", line[1:]))
-            elif line.rstrip("\n") == r"\ No newline at end of file":
-                continue
-            else:
-                die(f"bad hunk line in {path}: {line[:80].rstrip()}")
-        search = "".join(text for kind, text in sequence if kind in (" ", "-"))
-        replacement = "".join(text for kind, text in sequence if kind in (" ", "+"))
-        offset = old.find(search, search_from)
-        if offset < 0:
-            offset = old.find(search)
-        if offset < 0:
-            die(f"hunk context not found in {path}")
-        output.append(old[search_from:offset])
-        output.append(replacement)
-        search_from = offset + len(search)
-    output.append(old[search_from:])
-    write_file(path, "".join(output))
-
-
-def main():
-    if len(sys.argv) > 1 and sys.argv[1] in ("-h", "--help"):
-        print("apply_patch: read *** Begin Patch format from stdin; supports add/update/delete/move")
-        return
-    lines = sys.stdin.read().splitlines(keepends=True)
-    if not lines or lines[0].strip() != "*** Begin Patch":
-        die("patch must start with *** Begin Patch")
-    index = 1
-    while index < len(lines):
-        line = lines[index].rstrip("\n")
-        if line == "*** End Patch":
-            print("Done!")
-            return
-        if line.startswith("*** Add File: "):
-            path = line[len("*** Add File: "):].strip()
-            index += 1
-            content = []
-            while index < len(lines) and not lines[index].startswith("*** "):
-                if not lines[index].startswith("+"):
-                    die(f"add file lines must start with + for {path}")
-                content.append(lines[index][1:])
-                index += 1
-            if Path(path).exists():
-                die(f"file already exists: {path}")
-            write_file(path, "".join(content))
-            continue
-        if line.startswith("*** Delete File: "):
-            path = line[len("*** Delete File: "):].strip()
-            try:
-                Path(path).unlink()
-            except FileNotFoundError:
-                die(f"file not found: {path}")
-            index += 1
-            continue
-        if line.startswith("*** Update File: "):
-            path = line[len("*** Update File: "):].strip()
-            index += 1
-            move_to = None
-            if index < len(lines) and lines[index].startswith("*** Move to: "):
-                move_to = lines[index][len("*** Move to: "):].strip()
-                index += 1
-            body = []
-            while index < len(lines) and not (
-                lines[index].startswith("*** Add File: ")
-                or lines[index].startswith("*** Update File: ")
-                or lines[index].startswith("*** Delete File: ")
-                or lines[index].startswith("*** End Patch")
-            ):
-                if lines[index].startswith("*** Move to: "):
-                    die("move marker must appear before update hunks")
-                body.append(lines[index])
-                index += 1
-            if body:
-                apply_update(path, body)
-            elif move_to is None and not Path(path).exists():
-                die(f"file not found: {path}")
-            if move_to is not None:
-                move_file(path, move_to)
-            continue
-        die(f"unexpected patch line: {line}")
-    die("missing *** End Patch")
-
-
-if __name__ == "__main__":
-    main()
-`;
-
-const remoteShApplyPatchSource = String.raw`#!/bin/sh
+export const remoteApplyPatchSource = String.raw`#!/bin/sh
 set -eu
 
+allow_loose=0
+
 die() {
   printf 'apply_patch: %s\n' "$*" >&2
   exit 1
@@ -240,10 +99,16 @@ write_file() {
   cp "$source" "$target" || die "failed to write $target"
 }
 
+line_number_for_prefix() {
+  newline_count=$(printf '%s' "$1" | tr -cd '\n' | wc -c | tr -d ' ')
+  printf '%s\n' $((newline_count + 1))
+}
+
 replace_once() {
   target=$1
   search_file=$2
   replace_file=$3
+  hunk_id=$4
   [ -e "$target" ] || die "file not found: $target"
 
   marker="__UNIDESK_APPLY_PATCH_EOF_$$__"
@@ -255,16 +120,24 @@ replace_once() {
   replacement=${"$"}{replacement%"$marker"}
 
   if [ -z "$search" ]; then
+    [ "$allow_loose" = 1 ] || die "hunk $hunk_id in $target has no context; add unchanged/deleted anchor lines or pass --allow-loose after manual review"
+    printf 'apply_patch: hunk %s matched %s:1 (loose)\n' "$hunk_id" "$target" >&2
     new=$replacement$old
   else
     case "$old" in
       *"$search"*)
         prefix=${"$"}{old%%"$search"*}
         suffix=${"$"}{old#*"$search"}
+        case "$suffix" in
+          *"$search"*)
+            [ "$allow_loose" = 1 ] || die "hunk $hunk_id context matched multiple locations in $target; add more unchanged context or pass --allow-loose after manual review"
+            ;;
+        esac
+        printf 'apply_patch: hunk %s matched %s:%s\n' "$hunk_id" "$target" "$(line_number_for_prefix "$prefix")" >&2
         new=$prefix$replacement$suffix
         ;;
       *)
-        die "hunk context not found in $target"
+        die "hunk $hunk_id context not found in $target"
         ;;
     esac
   fi
@@ -282,10 +155,25 @@ apply_update() {
   search_file=
   replace_file=
   changed=0
+  hunk_number=0
+  has_add=0
+  has_delete=0
+  seen_add=0
+  anchor_before_add=0
+  anchor_after_add=0
+  explicit_eof=0
 
   finish_hunk() {
     [ "$in_hunk" = 1 ] || return 0
-    replace_once "$target" "$search_file" "$replace_file"
+    if [ "$allow_loose" != 1 ]; then
+      [ -s "$search_file" ] || die "hunk $hunk_number in $target has no context; add unchanged/deleted anchor lines or pass --allow-loose after manual review"
+      if [ "$has_add" = 1 ] && [ "$has_delete" = 0 ]; then
+        if [ "$anchor_before_add" != 1 ] || { [ "$anchor_after_add" != 1 ] && [ "$explicit_eof" != 1 ]; }; then
+          die "hunk $hunk_number in $target is insert-only without both leading and trailing context; include nearby unchanged lines after the insertion or pass --allow-loose after manual review"
+        fi
+      fi
+    fi
+    replace_once "$target" "$search_file" "$replace_file" "$hunk_number"
     rm -f "$search_file" "$replace_file"
     search_file=
     replace_file=
@@ -302,6 +190,13 @@ apply_update() {
         finish_hunk
         search_file=$(mk_temp)
         replace_file=$(mk_temp)
+        hunk_number=$((hunk_number + 1))
+        has_add=0
+        has_delete=0
+        seen_add=0
+        anchor_before_add=0
+        anchor_after_add=0
+        explicit_eof=0
         in_hunk=1
         continue
         ;;
@@ -312,17 +207,25 @@ apply_update() {
         text=${"$"}{line#?}
         printf '%s\n' "$text" >> "$search_file"
         printf '%s\n' "$text" >> "$replace_file"
+        if [ "$seen_add" = 1 ]; then anchor_after_add=1; else anchor_before_add=1; fi
         ;;
       "-"*)
         text=${"$"}{line#?}
         printf '%s\n' "$text" >> "$search_file"
+        has_delete=1
+        if [ "$seen_add" = 1 ]; then anchor_after_add=1; else anchor_before_add=1; fi
         ;;
       "+"*)
         text=${"$"}{line#?}
         printf '%s\n' "$text" >> "$replace_file"
+        has_add=1
+        seen_add=1
         ;;
       "\\ No newline at end of file")
         ;;
+      "*** End of File"*)
+        explicit_eof=1
+        ;;
       *)
         die "bad hunk line in $target: $line"
         ;;
@@ -432,10 +335,21 @@ parse_update_file() {
 }
 
 main() {
-  if [ "${"$"}{1:-}" = "-h" ] || [ "${"$"}{1:-}" = "--help" ]; then
-    printf 'apply_patch: sh-only helper; reads *** Begin Patch format from stdin\n'
-    return 0
-  fi
+  while [ "$#" -gt 0 ]; do
+    case "${"$"}1" in
+      -h|--help)
+        printf 'apply_patch: sh-only helper; reads *** Begin Patch format from stdin; --allow-loose bypasses low-context hunk guards\n'
+        return 0
+        ;;
+      --allow-loose)
+        allow_loose=1
+        shift
+        ;;
+      *)
+        die "unsupported option: ${"$"}1"
+        ;;
+    esac
+  done
   next_patch_line || die "patch must start with *** Begin Patch"
   [ "$line" = "*** Begin Patch" ] || die "patch must start with *** Begin Patch"
   while next_patch_line; do
@@ -1440,14 +1354,14 @@ function buildShellStdinCommand(args: string[]): string {
 function podApplyPatchStdinWrapper(): { prefix: string; suffix: string } {
   const toolMarker = "__UNIDESK_APPLY_PATCH_TOOL__";
   const patchMarker = "__UNIDESK_APPLY_PATCH_PAYLOAD__";
-  if (remoteShApplyPatchSource.includes(toolMarker)) throw new Error("remote apply_patch source contains reserved heredoc marker");
+  if (remoteApplyPatchSource.includes(toolMarker)) throw new Error("remote apply_patch source contains reserved heredoc marker");
   return {
     prefix: [
       "set -eu",
       'UNIDESK_SSH_TOOL_DIR="${UNIDESK_SSH_TOOL_DIR:-/tmp/unidesk-ssh-tools}"',
       'mkdir -p "$UNIDESK_SSH_TOOL_DIR"',
       `cat > "$UNIDESK_SSH_TOOL_DIR/apply_patch" <<'${toolMarker}'`,
-      remoteShApplyPatchSource.trimEnd(),
+      remoteApplyPatchSource.trimEnd(),
       toolMarker,
       'chmod 700 "$UNIDESK_SSH_TOOL_DIR/apply_patch"',
       `"$UNIDESK_SSH_TOOL_DIR/apply_patch" "$@" <<'${patchMarker}'`,

scripts/ssh-argv-guidance-contract-test.ts

@@ -1,6 +1,10 @@
+import { spawnSync } from "node:child_process";
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import os from "node:os";
+import path from "node:path";
 import { sshHelp } from "./src/help";
 import { providerTriageRecommendedCrossChecks } from "./src/provider-triage";
-import { formatSshFailureHint, parseSshArgs, parseSshInvocation, sshFailureHint } from "./src/ssh";
+import { formatSshFailureHint, parseSshArgs, parseSshInvocation, remoteApplyPatchSource, sshFailureHint } from "./src/ssh";
 
 type JsonRecord = Record<string, unknown>;
 
@@ -19,6 +23,35 @@ function assertThrows(fn: () => unknown, pattern: RegExp, message: string): void
   throw new Error(`${message}: expected throw`);
 }
 
+function applyPatchFixture(args: string[], patch: string, files: Record<string, string>): { status: number | null; stdout: string; stderr: string; files: Record<string, string> } {
+  const root = mkdtempSync(path.join(os.tmpdir(), "unidesk-apply-patch-contract-"));
+  try {
+    const helperPath = path.join(root, "apply_patch");
+    writeFileSync(helperPath, remoteApplyPatchSource, "utf8");
+    for (const [relativePath, content] of Object.entries(files)) {
+      const target = path.join(root, relativePath);
+      writeFileSync(target, content, "utf8");
+    }
+    const run = spawnSync("sh", [helperPath, ...args], {
+      cwd: root,
+      input: patch,
+      encoding: "utf8",
+    });
+    const outputFiles: Record<string, string> = {};
+    for (const relativePath of Object.keys(files)) {
+      outputFiles[relativePath] = readFileSync(path.join(root, relativePath), "utf8");
+    }
+    return {
+      status: run.status,
+      stdout: run.stdout,
+      stderr: run.stderr,
+      files: outputFiles,
+    };
+  } finally {
+    rmSync(root, { recursive: true, force: true });
+  }
+}
+
 export function runSshArgvGuidanceContract(): JsonRecord {
   const argv = parseSshArgs(["argv", "true"]);
   assertCondition(argv.invocationKind === "argv", "argv subcommand must be classified as argv", argv);
@@ -82,9 +115,99 @@ export function runSshArgvGuidanceContract(): JsonRecord {
   assertCondition(routeApplyPatch.parsed.requiresStdin === true, "k3s apply-patch operation must stream local patch stdin", routeApplyPatch);
   assertCondition(routeApplyPatch.parsed.remoteCommand === "'env' 'KUBECONFIG=/etc/rancher/k3s/k3s.yaml' 'kubectl' 'exec' '-i' '-n' 'hwlab-dev' 'deployment/hwlab-cloud-api' '--' 'sh' '-s' '--'", "D601:k3s:<namespace>:<workload> apply-patch must enter pod with stdin", routeApplyPatch);
   assertCondition(routeApplyPatch.parsed.stdinPrefix?.includes("apply_patch") && routeApplyPatch.parsed.stdinPrefix.includes("__UNIDESK_APPLY_PATCH_PAYLOAD__"), "k3s apply-patch operation must inject pod helper before patch stdin", routeApplyPatch);
+  assertCondition(routeApplyPatch.parsed.stdinPrefix?.includes("#!/bin/sh"), "k3s apply-patch operation must inject the same sh helper used by host apply-patch", routeApplyPatch);
   assertCondition(!routeApplyPatch.parsed.stdinPrefix?.includes("python3") && !routeApplyPatch.parsed.stdinPrefix?.includes("node "), "k3s apply-patch operation must use the sh-only pod helper", routeApplyPatch);
   assertCondition(routeApplyPatch.parsed.stdinSuffix === "\n__UNIDESK_APPLY_PATCH_PAYLOAD__\n", "k3s apply-patch operation must terminate patch heredoc", routeApplyPatch);
 
+  const hostApplyPatchLoose = parseSshArgs(["apply-patch", "--allow-loose"]);
+  assertCondition(hostApplyPatchLoose.remoteCommand === "'apply_patch' '--allow-loose'", "host apply-patch must pass --allow-loose as an explicit helper argument", hostApplyPatchLoose);
+
+  const safePatch = applyPatchFixture([], [
+    "*** Begin Patch",
+    "*** Update File: sample.txt",
+    "@@",
+    " function alpha() {",
+    "   return 1;",
+    " }",
+    "+function inserted() {",
+    "+  return 3;",
+    "+}",
+    " function beta() {",
+    "   return 2;",
+    " }",
+    "*** End Patch",
+    "",
+  ].join("\n"), {
+    "sample.txt": [
+      "function alpha() {",
+      "  return 1;",
+      "}",
+      "function beta() {",
+      "  return 2;",
+      "}",
+      "",
+    ].join("\n"),
+  });
+  assertCondition(safePatch.status === 0, "apply_patch sh helper should accept anchored insertions", safePatch);
+  assertCondition(safePatch.stderr.includes("apply_patch: hunk 1 matched sample.txt:1"), "apply_patch sh helper must report matched file:line", safePatch);
+  assertCondition(safePatch.files["sample.txt"]?.includes("function inserted()"), "apply_patch sh helper should apply anchored insertion", safePatch);
+
+  const lowContextPatch = applyPatchFixture([], [
+    "*** Begin Patch",
+    "*** Update File: sample.txt",
+    "@@",
+    " function alpha() {",
+    "   return 1;",
+    " }",
+    "+function misplaced() {",
+    "+  return 9;",
+    "+}",
+    "*** End Patch",
+    "",
+  ].join("\n"), {
+    "sample.txt": [
+      "function alpha() {",
+      "  return 1;",
+      "}",
+      "function beta() {",
+      "  return 2;",
+      "}",
+      "",
+    ].join("\n"),
+  });
+  assertCondition(lowContextPatch.status !== 0, "apply_patch sh helper should reject insert-only hunks without trailing context", lowContextPatch);
+  assertCondition(lowContextPatch.stderr.includes("insert-only without both leading and trailing context"), "low-context rejection must explain the risk", lowContextPatch);
+
+  const duplicateContextPatch = applyPatchFixture([], [
+    "*** Begin Patch",
+    "*** Update File: sample.txt",
+    "@@",
+    " marker",
+    "+inserted",
+    " x",
+    "*** End Patch",
+    "",
+  ].join("\n"), {
+    "sample.txt": "marker\nx\nmarker\nx\n",
+  });
+  assertCondition(duplicateContextPatch.status !== 0, "apply_patch sh helper should reject hunks whose context matches multiple locations", duplicateContextPatch);
+  assertCondition(duplicateContextPatch.stderr.includes("matched multiple locations"), "duplicate context rejection must explain the risk", duplicateContextPatch);
+
+  const reviewedLoosePatch = applyPatchFixture(["--allow-loose"], [
+    "*** Begin Patch",
+    "*** Update File: sample.txt",
+    "@@",
+    " marker",
+    "+reviewed",
+    " x",
+    "*** End Patch",
+    "",
+  ].join("\n"), {
+    "sample.txt": "marker\nx\nmarker\nx\n",
+  });
+  assertCondition(reviewedLoosePatch.status === 0, "--allow-loose should allow an explicitly reviewed ambiguous hunk", reviewedLoosePatch);
+  assertCondition(reviewedLoosePatch.files["sample.txt"] === "marker\nreviewed\nx\nmarker\nx\n", "--allow-loose should still apply only one hunk", reviewedLoosePatch);
+
   assertThrows(
     () => parseSshInvocation("D601:k3s:kubectl", ["get", "pods"]),
     /route must locate a target only.*ssh D601:k3s kubectl/u,
@@ -143,6 +266,7 @@ export function runSshArgvGuidanceContract(): JsonRecord {
   assertCondition(helpText.includes("ssh D601:k3s:hwlab-dev:hwlab-cloud-api/app pwd"), "ssh help must document k3s pod workspace route", helpText);
   assertCondition(helpText.includes("ssh D601:k3s script <<'SCRIPT'"), "ssh help must document k3s control-plane script operation", helpText);
   assertCondition(helpText.includes("ssh D601:k3s:hwlab-dev:hwlab-cloud-api apply-patch <<'PATCH'"), "ssh help must document k3s pod apply-patch operation", helpText);
+  assertCondition(helpText.includes("apply-patch [--allow-loose]") && helpText.includes("low-context update hunks"), "ssh help must document apply-patch loose-context guard", helpText);
   assertCondition(helpText.includes("ssh D601:k3s:hwlab-dev:hwlab-cloud-api script <<'SCRIPT'"), "ssh help must document k3s script operation", helpText);
   assertCondition(helpText.includes("UNIDESK_SSH_HINT"), "ssh help must document structured failure hint", helpText);
 
@@ -155,6 +279,7 @@ export function runSshArgvGuidanceContract(): JsonRecord {
       "argv form is classified and quoted as the success path for non-interactive commands",
       "stdin script form removes shell-command strings for host and k3s workload scripts",
       "pod apply-patch operation injects helper and forwards patch stdin",
+      "apply-patch uses one sh helper for host and pod paths and rejects low-context hunks unless --allow-loose is explicit",
       "legacy operation-in-route forms are rejected in any k3s route segment with canonical route-plus-operation guidance",
       "post-provider k3s shorthand is rejected so location and operation stay separated",
       "k3s route stays location-only while operations fix native kubeconfig and assemble kubectl exec as argv",