Merge pull request #2066 from pikasTech/fix/2060-pikaoa-test-target-async

修复 PikaOA 测试启动长连接阻塞

scripts/src/pikaoa-test-target-async.test.ts

@@ -0,0 +1,247 @@
+import assert from "node:assert/strict";
+import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { test } from "bun:test";
+
+import type { UniDeskConfig } from "./config";
+import { rootPath } from "./config";
+import { runPikaoaCommand } from "./pikaoa-test-target";
+import type { SshCaptureResult } from "./ssh";
+
+test("PikaOA test target submits once and exposes bounded typed status", async () => {
+  const root = mkdtempSync(join(tmpdir(), "pikaoa-test-target-"));
+  const stateRoot = join(root, "state");
+  const fakeRoot = join(root, "fake-k8s");
+  const binDir = join(root, "bin");
+  const secretsDir = join(root, "secrets");
+  mkdirSync(binDir, { recursive: true });
+  mkdirSync(secretsDir, { recursive: true });
+  mkdirSync(join(fakeRoot, "namespaces"), { recursive: true });
+
+  const secretValues = ["fixture-db-password", "fixture-admin-password", "fixture-employee-password", "fixture-session-secret"];
+  for (const [name, value] of [
+    ["pikaoa-test-database-password.txt", secretValues[0]],
+    ["pikaoa-test-admin-password.txt", secretValues[1]],
+    ["pikaoa-test-employee-password.txt", secretValues[2]],
+    ["pikaoa-test-session-secret.txt", secretValues[3]],
+  ]) writeFileSync(join(secretsDir, name), `${value}\n`, { mode: 0o600 });
+
+  const fakeKubectl = join(binDir, "kubectl");
+  writeFileSync(fakeKubectl, `#!/bin/sh
+set -u
+root="$FAKE_K8S_ROOT"
+args="$*"
+namespace=""
+previous=""
+for argument in "$@"; do
+  if [ "$previous" = "-n" ]; then namespace="$argument"; fi
+  previous="$argument"
+done
+namespace_after_resource() {
+  previous=""
+  for argument in "$@"; do
+    if [ "$previous" = "namespace" ]; then printf '%s' "$argument"; return; fi
+    previous="$argument"
+  done
+}
+if printf '%s' "$args" | grep -Eq '(^| )apply( |$)'; then
+  file=""
+  previous=""
+  for argument in "$@"; do
+    if [ "$previous" = "-f" ]; then file="$argument"; fi
+    previous="$argument"
+  done
+  if grep -q '"kind":"Namespace"' "$file"; then
+    applied_namespace="$(python3 - "$file" <<'PY'
+import json, pathlib, sys
+for line in pathlib.Path(sys.argv[1]).read_text().splitlines():
+    if not line or line == "---":
+        continue
+    item = json.loads(line)
+    if item.get("kind") == "Namespace":
+        print(item["metadata"]["name"])
+        break
+PY
+)"
+    : >"$root/namespaces/$applied_namespace"
+  fi
+  [ ! -f "$root/slow" ] || sleep 0.2
+  exit 0
+fi
+if printf '%s' "$args" | grep -q ' rollout status '; then
+  [ ! -f "$root/slow" ] || sleep 0.4
+  exit 0
+fi
+if printf '%s' "$args" | grep -q ' wait '; then
+  [ ! -f "$root/slow" ] || sleep 0.4
+  if [ -f "$root/fail-migration" ]; then printf '%s\n' 'migration failed' >&2; exit 1; fi
+  exit 0
+fi
+if printf '%s' "$args" | grep -Eq '(^| )get namespace '; then
+  name="$(namespace_after_resource "$@")"
+  if [ -f "$root/fail-runtime-query" ]; then
+    printf '%s\n' 'unable to connect to the server' >&2
+    exit 1
+  fi
+  if [ ! -f "$root/namespaces/$name" ]; then
+    printf 'Error from server (NotFound): namespaces "%s" not found\n' "$name" >&2
+    exit 1
+  fi
+  instance="\${name#pikaoa-test-}"
+  if printf '%s' "$args" | grep -q 'managed-by'; then printf '%s' 'unidesk-pikaoa-test-target'; exit 0; fi
+  if printf '%s' "$args" | grep -q 'unidesk.*target'; then printf '%s' 'TEST01'; exit 0; fi
+  if printf '%s' "$args" | grep -q 'unidesk.*instance'; then printf '%s' "$instance"; exit 0; fi
+  printf '{"metadata":{"name":"%s","labels":{"app.kubernetes.io/managed-by":"unidesk-pikaoa-test-target","pikaoa.unidesk.io/target":"TEST01","pikaoa.unidesk.io/instance":"%s"}}}\n' "$name" "$instance"
+  exit 0
+fi
+if printf '%s' "$args" | grep -q ' get deployment,statefulset,job,service,persistentvolumeclaim,pod '; then
+  if [ -f "$root/fail-runtime-resource-query" ]; then
+    printf '%s\n' 'resource query failed' >&2
+    exit 1
+  fi
+  printf '%s\n' '{"items":[]}'
+  exit 0
+fi
+if printf '%s' "$args" | grep -Eq '(^| )delete namespace '; then
+  name="$(namespace_after_resource "$@")"
+  rm -f "$root/namespaces/$name"
+  exit 0
+fi
+printf 'unsupported fake kubectl call: %s\n' "$args" >&2
+exit 64
+`, { mode: 0o700 });
+  chmodSync(fakeKubectl, 0o700);
+
+  const fixturePath = join(root, "pikaoa.yaml");
+  const fixture = readFileSync(rootPath("config", "fixtures", "pikaoa-test-target.yaml"), "utf8")
+    .replace("validationOnly: true", "validationOnly: false")
+    .replace("statusRequestTimeoutSeconds: 3", "statusRequestTimeoutSeconds: 31")
+    .replace("logTailLines: 8", "logTailLines: 101")
+    .replace("/tmp/unidesk-pikaoa-test-target-fixture", stateRoot);
+  writeFileSync(fixturePath, fixture);
+
+  const config = {} as UniDeskConfig;
+  let lastCapture: SshCaptureResult | null = null;
+  const localCapture = async (_config: UniDeskConfig, _route: string, args: string[], stdin: string): Promise<SshCaptureResult> => {
+    assert.deepEqual(args, ["sh"]);
+    const child = Bun.spawn(["sh"], {
+      env: { ...process.env, PATH: `${binDir}:${process.env.PATH ?? ""}`, FAKE_K8S_ROOT: fakeRoot },
+      stdin: "pipe",
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    child.stdin.write(stdin);
+    child.stdin.end();
+    const [stdout, stderr, exitCode] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]);
+    lastCapture = { stdout, stderr, exitCode };
+    return lastCapture;
+  };
+  const args = (action: string, instance: string, withCommit = false): string[] => [
+    "test-target", action, "--config", fixturePath, "--target", "TEST01", "--instance", instance,
+    ...(withCommit ? ["--commit", "0123456789abcdef"] : []),
+    ...((action === "start" || action === "stop") ? ["--confirm"] : []),
+    "--output", "json",
+  ];
+  const projection = async (action: string, instance: string, withCommit = false): Promise<Record<string, unknown>> => {
+    const result = await runPikaoaCommand(config, args(action, instance, withCommit), localCapture);
+    return JSON.parse(result.renderedText) as Record<string, unknown>;
+  };
+  const waitFor = async (instance: string, expectedCode: string): Promise<Record<string, unknown>> => {
+    for (let attempt = 0; attempt < 80; attempt += 1) {
+      const value = await projection("status", instance);
+      const status = value.status as Record<string, unknown>;
+      if (status.code === expectedCode) return value;
+      await Bun.sleep(25);
+    }
+    assert.fail(`status did not reach ${expectedCode}`);
+  };
+
+  try {
+    const missing = await projection("status", "missing");
+    assert.equal((missing.status as Record<string, unknown>).code, "start-not-found");
+
+    writeFileSync(join(fakeRoot, "slow"), "1\n");
+    const submitted = await projection("start", "success", true);
+    assert.equal(submitted.code, "start-submitted", JSON.stringify(lastCapture));
+    assert.equal(submitted.mutation, true);
+    assert.deepEqual((submitted.target as Record<string, unknown>).taskState, {
+      rootPath: stateRoot,
+      statusRequestTimeoutSeconds: 31,
+      logTailLines: 101,
+    });
+    const running = await projection("status", "success");
+    assert.equal((running.status as Record<string, unknown>).code, "start-running");
+    rmSync(join(fakeRoot, "slow"));
+    const succeeded = await waitFor("success", "start-succeeded");
+    const succeededStatus = succeeded.status as Record<string, unknown>;
+    assert.equal((succeededStatus.task as Record<string, unknown>).exitCode, 0);
+    assert.ok((succeededStatus.logTail as unknown[]).length <= 101);
+
+    writeFileSync(join(fakeRoot, "fail-runtime-query"), "1\n");
+    const runtimeQueryFailed = await projection("status", "success");
+    const runtimeQueryFailedStatus = runtimeQueryFailed.status as Record<string, unknown>;
+    assert.equal(runtimeQueryFailed.ok, false);
+    assert.equal(runtimeQueryFailedStatus.code, "runtime-query-failed");
+    assert.equal(runtimeQueryFailedStatus.taskCode, "start-succeeded");
+    assert.equal((runtimeQueryFailedStatus.runtime as Record<string, unknown>).code, "runtime-query-failed");
+    const runtimeQueryFailedText = await runPikaoaCommand(config, args("status", "success").slice(0, -2), localCapture);
+    assert.match(runtimeQueryFailedText.renderedText, /status=runtime-query-failed/);
+    rmSync(join(fakeRoot, "fail-runtime-query"));
+
+    writeFileSync(join(fakeRoot, "fail-runtime-resource-query"), "1\n");
+    const runtimeResourceQueryFailed = await projection("status", "success");
+    const runtimeResourceQueryFailedStatus = runtimeResourceQueryFailed.status as Record<string, unknown>;
+    assert.equal(runtimeResourceQueryFailed.ok, false);
+    assert.equal(runtimeResourceQueryFailedStatus.code, "runtime-resource-query-failed");
+    assert.equal(runtimeResourceQueryFailedStatus.taskCode, "start-succeeded");
+    assert.equal((runtimeResourceQueryFailedStatus.runtime as Record<string, unknown>).code, "runtime-resource-query-failed");
+    rmSync(join(fakeRoot, "fail-runtime-resource-query"));
+
+    const repeated = await projection("start", "success", true);
+    assert.equal(repeated.code, "start-already-succeeded");
+    assert.equal(repeated.mutation, false);
+
+    writeFileSync(join(fakeRoot, "fail-migration"), "1\n");
+    await projection("start", "failed", true);
+    const failed = await waitFor("failed", "start-failed");
+    const failedTask = ((failed.status as Record<string, unknown>).task as Record<string, unknown>);
+    assert.equal(failedTask.code, "migration-wait-failed");
+    assert.equal(failedTask.exitCode, 46);
+    rmSync(join(fakeRoot, "fail-migration"));
+
+    writeFileSync(join(fakeRoot, "slow"), "1\n");
+    await projection("start", "cancel", true);
+    const stopped = await projection("stop", "cancel");
+    assert.equal(stopped.startTaskDisposition, "cancel-requested");
+    assert.equal(stopped.mutation, true);
+    await waitFor("cancel", "start-canceled");
+
+    const serialized = JSON.stringify({ submitted, succeeded, failed, stopped });
+    for (const value of secretValues) assert.equal(serialized.includes(value), false);
+    assert.equal(serialized.includes("postgres://pikaoa_test:"), false);
+  } finally {
+    rmSync(root, { recursive: true, force: true });
+  }
+}, 15_000);
+
+test("validation-only fixture never invokes remote capture", async () => {
+  let calls = 0;
+  const forbiddenCapture = async (): Promise<SshCaptureResult> => {
+    calls += 1;
+    throw new Error("validation-only fixture attempted remote capture");
+  };
+  const result = await runPikaoaCommand({} as UniDeskConfig, [
+    "test-target", "status", "--config", "config/fixtures/pikaoa-test-target.yaml", "--target", "TEST01", "--instance", "validation-only", "--output", "json",
+  ], forbiddenCapture);
+  assert.equal(calls, 0);
+  const payload = JSON.parse(result.renderedText) as Record<string, unknown>;
+  assert.equal(payload.remoteQueried, false);
+  assert.equal((payload.status as Record<string, unknown>).remoteQueried, false);
+
+  const unconfigured = await runPikaoaCommand({} as UniDeskConfig, ["test-target", "status", "--output", "json"], forbiddenCapture);
+  const unconfiguredPayload = JSON.parse(unconfigured.renderedText) as Record<string, unknown>;
+  assert.equal(calls, 0);
+  assert.equal(unconfiguredPayload.configured, false);
+  assert.equal(unconfiguredPayload.remoteQueried, false);
+});

scripts/src/pikaoa-test-target.ts

@@ -39,6 +39,11 @@ interface TestTargetSpec {
   ttlSeconds: number;
   waitTimeoutSeconds: number;
   fieldManager: string;
+  taskState: {
+    rootPath: string;
+    statusRequestTimeoutSeconds: number;
+    logTailLines: number;
+  };
   cleanup: {
     mode: "delete-namespace";
     requireOwnershipLabels: true;
@@ -141,6 +146,13 @@ interface SecretMaterial {
   employeePassword: string;
 }
 
+type RemoteCapture = typeof capture;
+
+interface AsyncJobIdentity {
+  id: string;
+  stateDir: string;
+}
+
 const DEFAULT_CONFIG_PATH = rootPath("config", "pikaoa.yaml");
 const MANAGED_BY = "unidesk-pikaoa-test-target";
 const TEST_RUNTIME_LABEL = "pikaoa.unidesk.io/test-runtime";
@@ -161,21 +173,22 @@ export function pikaoaTestTargetHelp(): Record<string, unknown> {
     guarantees: [
       "未声明专用 target 时 plan/status 返回 configured=false，start/stop 在远端调用前失败。",
       "validationOnly target 只用于本地渲染，不连接 route。",
+      "start 只提交具名远程任务并立即返回；status 用短连接读取阶段、终态和有界事件尾部。",
       "正式 delivery namespace、PaC、GitOps、公网暴露和生产数据库不属于该入口的写入范围。",
       "Secret 只披露 sourceRef、targetKey、presence 和 fingerprint，不输出值。",
     ],
   };
 }
 
-export async function runPikaoaCommand(config: UniDeskConfig, args: string[]): Promise<RenderedCliResult> {
+export async function runPikaoaCommand(config: UniDeskConfig, args: string[], remoteCapture: RemoteCapture = capture): Promise<RenderedCliResult> {
   if (args.length === 0 || args.some(isHelpToken)) return renderMachine("pikaoa test-target", pikaoaTestTargetHelp(), "json");
   if (args[0] !== "test-target") throw inputError("pikaoa 只支持 test-target", "unsupported-pikaoa-command", "pikaoa", ["test-target"]);
   const options = parseOptions(args.slice(1));
   const selection = readSelection(options);
   if (options.action === "plan") return renderResult(planPayload(options, selection), options);
-  if (options.action === "status") return await statusResult(config, options, selection);
-  if (options.action === "start") return await startResult(config, options, selection);
-  return await stopResult(config, options, selection);
+  if (options.action === "status") return await statusResult(config, options, selection, remoteCapture);
+  if (options.action === "start") return await startResult(config, options, selection, remoteCapture);
+  return await stopResult(config, options, selection, remoteCapture);
 }
 
 function parseOptions(args: string[]): TestTargetOptions {
@@ -269,6 +282,7 @@ function parseTarget(id: string, root: Record<string, unknown>, configLabel: str
   const workerProbes = record(probes.worker, `${path}.probes.worker`);
   const cleanup = record(root.cleanup, `${path}.cleanup`);
   const migration = record(root.migration, `${path}.migration`);
+  const taskState = record(root.taskState, `${path}.taskState`);
   if (exporterFailure.blocking !== false) throw new Error(`${path}.observability.exporterFailure.blocking 必须为 false`);
   if (exporterFailure.warning !== true) throw new Error(`${path}.observability.exporterFailure.warning 必须为 true`);
   if (cleanup.requireOwnershipLabels !== true) throw new Error(`${path}.cleanup.requireOwnershipLabels 必须为 true`);
@@ -283,6 +297,11 @@ function parseTarget(id: string, root: Record<string, unknown>, configLabel: str
     ttlSeconds: positiveInteger(root.ttlSeconds, `${path}.ttlSeconds`, 86_400),
     waitTimeoutSeconds: positiveInteger(root.waitTimeoutSeconds, `${path}.waitTimeoutSeconds`, 900),
     fieldManager: kubernetesName(nonEmpty(root.fieldManager, `${path}.fieldManager`), `${path}.fieldManager`, 63),
+    taskState: {
+      rootPath: taskStateRoot(taskState.rootPath, `${path}.taskState.rootPath`),
+      statusRequestTimeoutSeconds: positiveSafeInteger(taskState.statusRequestTimeoutSeconds, `${path}.taskState.statusRequestTimeoutSeconds`),
+      logTailLines: positiveSafeInteger(taskState.logTailLines, `${path}.taskState.logTailLines`),
+    },
     cleanup: {
       mode: exactString(cleanup.mode, `${path}.cleanup.mode`, "delete-namespace") as "delete-namespace",
       requireOwnershipLabels: true,
@@ -451,16 +470,25 @@ function renderedPlan(selection: Selection, context: RenderContext): Record<stri
       command: target.migration.command,
       applyPhase: "after-postgres-before-workloads",
     },
+    taskState: {
+      jobId: asyncJobIdentity(target, context).id,
+      rootPath: target.taskState.rootPath,
+      statusRequestTimeoutSeconds: target.taskState.statusRequestTimeoutSeconds,
+      logTailLines: target.taskState.logTailLines,
+      mutation: false,
+    },
     manifestFingerprint: structuralFingerprint(structuralManifest),
     mutation: false,
   };
 }
 
-async function statusResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection): Promise<RenderedCliResult> {
+async function statusResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection, remoteCapture: RemoteCapture): Promise<RenderedCliResult> {
   const planned = planPayload(options, selection);
-  if (selection.target === null) return renderResult(planned, options);
+  if (selection.target === null) {
+    return renderResult({ ...planned, remoteQueried: false, status: { remoteQueried: false, reason: "test-target-not-configured" } }, options);
+  }
   if (selection.target.validationOnly) {
-    return renderResult({ ...planned, status: { remoteQueried: false, reason: "validation-only-target" } }, options);
+    return renderResult({ ...planned, remoteQueried: false, status: { remoteQueried: false, reason: "validation-only-target" } }, options);
   }
   const context = renderContext(options, selection.target, false);
   if (context === null) return renderResult({ ...planned, ok: false, status: { remoteQueried: false, reason: "instance-required" } }, options);
@@ -468,13 +496,13 @@ async function statusResult(config: UniDeskConfig, options: TestTargetOptions, s
   if (blockers.length > 0) {
     return renderResult({ ...planned, ok: false, mutation: false, remoteQueried: false, blockers, status: { remoteQueried: false, reason: "preflight-blocked" } }, options);
   }
-  const remote = await capture(config, selection.target.route, ["sh"], statusScript(selection.target, context));
+  const remote = await remoteCapture(config, selection.target.route, ["sh"], statusScript(selection.target, context));
   const parsed = parseJsonOutput(remote.stdout);
   const status = parsed === null ? { ok: false, remote: compactCapture(remote, { full: true }) } : summarizeRemoteStatus(parsed, selection.target, context);
-  return renderResult({ ...planned, ok: remote.exitCode === 0 && status.ok !== false, status: { remoteQueried: true, ...status } }, options);
+  return renderResult({ ...planned, ok: remote.exitCode === 0 && status.ok !== false, remoteQueried: true, status: { remoteQueried: true, ...status } }, options);
 }
 
-async function startResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection): Promise<RenderedCliResult> {
+async function startResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection, remoteCapture: RemoteCapture): Promise<RenderedCliResult> {
   const planned = planPayload(options, selection);
   const context = selection.target === null ? null : renderContext(options, selection.target);
   const blockers = safetyBlockers(options, selection, context);
@@ -485,24 +513,28 @@ async function startResult(config: UniDeskConfig, options: TestTargetOptions, se
   }
   const material = readSecretMaterial(selection, selection.target);
   if (!material.ok) {
-    return renderResult({ ...planned, ok: false, mutation: false, blockers: material.blockers, secret: { sources: material.sources, valuesPrinted: false }, failure: "secret-preflight-blocked" }, options);
+    return renderResult({ ...planned, ok: false, mutation: false, remoteQueried: false, blockers: material.blockers, secret: { sources: material.sources, valuesPrinted: false }, failure: "secret-preflight-blocked" }, options);
   }
   const manifest = buildManifest(selection.target, context, material.values);
-  const remote = await capture(config, selection.target.route, ["sh"], startScript(selection.target, context, manifest));
+  const remote = await remoteCapture(config, selection.target.route, ["sh"], startScript(selection.target, context, manifest));
   const parsed = parseJsonOutput(remote.stdout);
   const mutation = parsed?.mutation === true;
   return renderResult({
     ...planned,
     ok: remote.exitCode === 0 && parsed?.ok === true,
     mutation,
-    mode: mutation ? "confirmed" : "preflight-blocked",
+    remoteQueried: true,
+    mode: mutation ? "submitted" : parsed === null ? "submit-failed" : "existing-task",
+    code: parsed?.code ?? "start-submit-unreadable",
+    jobId: parsed?.jobId ?? asyncJobIdentity(selection.target, context).id,
+    task: parsed?.task ?? null,
     secret: { sources: material.sources, valuesPrinted: false },
     remote: parsed ?? compactCapture(remote, { full: true }),
     manifestFingerprint: structuralFingerprint(manifest),
   }, options);
 }
 
-async function stopResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection): Promise<RenderedCliResult> {
+async function stopResult(config: UniDeskConfig, options: TestTargetOptions, selection: Selection, remoteCapture: RemoteCapture): Promise<RenderedCliResult> {
   const planned = planPayload(options, selection);
   const context = selection.target === null ? null : renderContext(options, selection.target, false);
   const blockers = safetyBlockers(options, selection, context, false);
@@ -511,13 +543,17 @@ async function stopResult(config: UniDeskConfig, options: TestTargetOptions, sel
   if (selection.target === null || context === null || blockers.length > 0) {
     return renderResult({ ...planned, ok: false, mutation: false, remoteQueried: false, blockers: uniqueBlockers(blockers), failure: "preflight-blocked" }, options);
   }
-  const remote = await capture(config, selection.target.route, ["sh"], stopScript(selection.target, context));
+  const remote = await remoteCapture(config, selection.target.route, ["sh"], stopScript(selection.target, context));
   const parsed = parseJsonOutput(remote.stdout);
   return renderResult({
     ...planned,
     ok: remote.exitCode === 0 && parsed?.ok === true,
     mutation: parsed?.mutation === true,
+    remoteQueried: true,
     mode: parsed?.mutation === true ? "confirmed" : "no-change",
+    code: parsed?.code ?? "stop-result-unreadable",
+    jobId: parsed?.jobId ?? asyncJobIdentity(selection.target, context).id,
+    startTaskDisposition: parsed?.startTaskDisposition ?? "unknown",
     remote: parsed ?? compactCapture(remote, { full: true }),
   }, options);
 }
@@ -538,6 +574,12 @@ function renderContext(options: TestTargetOptions, target: TestTargetSpec, requi
   };
 }
 
+function asyncJobIdentity(target: TestTargetSpec, context: RenderContext): AsyncJobIdentity {
+  const digest = createHash("sha256").update(target.id).update("\0").update(context.instanceId).digest("hex").slice(0, 20);
+  const id = `pikaoa-${digest}`;
+  return { id, stateDir: `${target.taskState.rootPath}/${id}` };
+}
+
 function safetyBlockers(options: TestTargetOptions, selection: Selection, context: RenderContext | null, requireCommit = true): Blocker[] {
   const blockers: Blocker[] = [];
   const target = selection.target;
@@ -771,6 +813,117 @@ function readSecretMaterial(selection: Selection, target: TestTargetSpec): {
 }
 
 function startScript(target: TestTargetSpec, context: RenderContext, manifest: Record<string, unknown>[]): string {
+  const job = asyncJobIdentity(target, context);
+  const requestId = createHash("sha256")
+    .update(target.id)
+    .update("\0")
+    .update(context.instanceId)
+    .update("\0")
+    .update(context.commit)
+    .digest("hex");
+  const runnerEncoded = Buffer.from(startRunnerScript(target, context, manifest, job), "utf8").toString("base64");
+  return `
+set -u
+umask 077
+state_root=${shQuote(target.taskState.rootPath)}
+state_dir=${shQuote(job.stateDir)}
+status_file="$state_dir/status.json"
+request_file="$state_dir/request-id"
+pid_file="$state_dir/pid"
+job_id=${shQuote(job.id)}
+request_id=${shQuote(requestId)}
+emit_existing() {
+  python3 - "$request_file" "$status_file" "$pid_file" "$job_id" "$request_id" <<'PY'
+import json, os, pathlib, sys
+request_path, status_path, pid_path = map(pathlib.Path, sys.argv[1:4])
+job_id, expected_request = sys.argv[4:6]
+actual_request = request_path.read_text(errors="replace").strip() if request_path.exists() else ""
+if actual_request != expected_request:
+    print(json.dumps({"ok": False, "mutation": False, "code": "start-instance-conflict", "jobId": job_id, "submitted": False, "task": None, "valuesPrinted": False}))
+    raise SystemExit(0)
+try:
+    task = json.loads(status_path.read_text())
+except Exception:
+    print(json.dumps({"ok": False, "mutation": False, "code": "start-state-unreadable", "jobId": job_id, "submitted": False, "task": None, "valuesPrinted": False}))
+    raise SystemExit(0)
+state = str(task.get("state") or "unknown")
+runner_alive = False
+pid = task.get("pid")
+if isinstance(pid, int) and pid > 0:
+    try:
+        os.kill(pid, 0)
+        runner_alive = True
+    except OSError:
+        pass
+code_by_state = {
+    "queued": "start-already-running",
+    "running": "start-already-running",
+    "succeeded": "start-already-succeeded",
+    "failed": "start-already-failed",
+    "canceled": "start-already-canceled",
+}
+code = "start-worker-missing" if state in ("queued", "running") and not runner_alive else code_by_state.get(state, "start-state-unknown")
+ok = state not in ("failed", "canceled", "unknown") and code != "start-worker-missing"
+print(json.dumps({"ok": ok, "mutation": False, "code": code, "jobId": job_id, "submitted": False, "task": task, "valuesPrinted": False}))
+PY
+}
+if ! mkdir -p "$state_root"; then
+  printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-root-create-failed","valuesPrinted":false}'
+  exit 41
+fi
+if ! mkdir "$state_dir" 2>/dev/null; then
+  if [ -d "$state_dir" ]; then emit_existing; exit 0; fi
+  printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-create-failed","valuesPrinted":false}'
+  exit 42
+fi
+started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+printf '%s\n' "$request_id" >"$request_file"
+printf '%s\n' "$started_at" >"$state_dir/started-at"
+if ! printf '%s' ${shQuote(runnerEncoded)} | base64 -d >"$state_dir/job.sh"; then
+  rm -rf "$state_dir"
+  printf '%s\n' '{"ok":false,"mutation":false,"code":"task-runner-write-failed","valuesPrinted":false}'
+  exit 43
+fi
+chmod 0700 "$state_dir/job.sh"
+python3 - "$status_file" "$job_id" ${shQuote(target.id)} ${shQuote(context.instanceId)} ${shQuote(context.namespace)} ${shQuote(context.commit)} "$started_at" <<'PY'
+import json, os, pathlib, sys
+path = pathlib.Path(sys.argv[1])
+payload = {
+    "ok": True, "jobId": sys.argv[2], "targetId": sys.argv[3], "instanceId": sys.argv[4],
+    "namespace": sys.argv[5], "sourceCommit": sys.argv[6], "state": "running", "phase": "submitted",
+    "code": "start-running", "startedAt": sys.argv[7], "updatedAt": sys.argv[7], "finishedAt": None,
+    "exitCode": None, "pid": None, "valuesPrinted": False,
+}
+tmp = path.with_name(path.name + ".tmp")
+tmp.write_text(json.dumps(payload, separators=(",", ":")) + "\\n")
+os.replace(tmp, path)
+PY
+if [ "$?" -ne 0 ]; then
+  rm -rf "$state_dir"
+  printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-initialize-failed","valuesPrinted":false}'
+  exit 44
+fi
+nohup sh "$state_dir/job.sh" >/dev/null 2>&1 </dev/null &
+pid=$!
+printf '%s\n' "$pid" >"$pid_file"
+python3 - "$status_file" "$pid" <<'PY'
+import json, os, pathlib, sys
+path = pathlib.Path(sys.argv[1])
+try:
+    payload = json.loads(path.read_text())
+except Exception:
+    raise SystemExit(0)
+if payload.get("pid") is None and payload.get("phase") == "submitted":
+    payload["pid"] = int(sys.argv[2])
+    tmp = path.with_name(path.name + ".tmp.submit")
+    tmp.write_text(json.dumps(payload, separators=(",", ":")) + "\\n")
+    os.replace(tmp, path)
+PY
+printf '{"ok":true,"mutation":true,"code":"start-submitted","submitted":true,"jobId":"%s","pid":%s,"task":{"state":"running","phase":"submitted","code":"start-running"},"valuesPrinted":false}\n' "$job_id" "$pid"
+`;
+}
+
+function startRunnerScript(target: TestTargetSpec, context: RenderContext, manifest: Record<string, unknown>[], job: AsyncJobIdentity): string {
   const migration = manifest.filter((object) => object.kind === "Job");
   const workloads = manifest.filter((object) => object.kind === "Deployment");
   const foundation = manifest.filter((object) => object.kind !== "Job" && object.kind !== "Deployment");
@@ -778,132 +931,328 @@ function startScript(target: TestTargetSpec, context: RenderContext, manifest: R
   const foundationEncoded = encode(foundation);
   const migrationEncoded = encode(migration);
   const workloadsEncoded = encode(workloads);
-  return `
-set -u
+  return `#!/bin/sh
+set -eu
+umask 077
+state_dir=${shQuote(job.stateDir)}
+status_file="$state_dir/status.json"
+events_file="$state_dir/events.ndjson"
+started_at="$(cat "$state_dir/started-at")"
+job_id=${shQuote(job.id)}
 namespace=${shQuote(context.namespace)}
 managed_by=${shQuote(MANAGED_BY)}
 target_id=${shQuote(target.id)}
 instance_id=${shQuote(context.instanceId)}
+source_commit=${shQuote(context.commit)}
+current_phase=runner-start
+failure_code=start-runner-failed
+tmp=""
+write_status() {
+  state="$1"; phase="$2"; code="$3"; exit_code="$4"
+  now="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+  python3 - "$status_file" "$events_file" "$job_id" "$target_id" "$instance_id" "$namespace" "$source_commit" "$state" "$phase" "$code" "$started_at" "$now" "$exit_code" "$$" <<'PY'
+import json, os, pathlib, sys
+status_path = pathlib.Path(sys.argv[1])
+events_path = pathlib.Path(sys.argv[2])
+state, phase, code = sys.argv[8:11]
+exit_code = None if sys.argv[13] == "null" else int(sys.argv[13])
+terminal = state in ("succeeded", "failed", "canceled")
+payload = {
+    "ok": state in ("running", "succeeded"), "jobId": sys.argv[3], "targetId": sys.argv[4],
+    "instanceId": sys.argv[5], "namespace": sys.argv[6], "sourceCommit": sys.argv[7],
+    "state": state, "phase": phase, "code": code, "startedAt": sys.argv[11], "updatedAt": sys.argv[12],
+    "finishedAt": sys.argv[12] if terminal else None, "exitCode": exit_code, "pid": int(sys.argv[14]),
+    "valuesPrinted": False,
+}
+tmp = status_path.with_name(status_path.name + ".tmp." + sys.argv[14])
+tmp.write_text(json.dumps(payload, separators=(",", ":")) + "\\n")
+os.replace(tmp, status_path)
+event = {"at": sys.argv[12], "state": state, "phase": phase, "code": code, "valuesPrinted": False}
+with events_path.open("a", encoding="utf-8") as handle:
+    handle.write(json.dumps(event, separators=(",", ":")) + "\\n")
+PY
+}
+check_canceled() {
+  if [ -f "$state_dir/cancel-requested" ]; then failure_code=start-canceled; exit 130; fi
+}
+finish() {
+  rc=$?
+  trap - EXIT HUP INT TERM
+  if [ -f "$state_dir/cancel-requested" ]; then
+    write_status canceled canceled start-canceled "$rc" || true
+  elif [ "$rc" -eq 0 ]; then
+    write_status succeeded completed started 0 || true
+  else
+    write_status failed "$current_phase" "$failure_code" "$rc" || true
+  fi
+  [ -z "$tmp" ] || rm -rf "$tmp"
+  exit "$rc"
+}
+trap finish EXIT
+trap 'failure_code=start-runner-hangup; exit 129' HUP
+trap 'failure_code=start-runner-interrupted; exit 130' INT
+trap 'failure_code=start-runner-terminated; exit 143' TERM
+rm -f "$state_dir/job.sh"
 tmp="$(mktemp -d)"
-trap 'rm -rf "$tmp"' EXIT
-if kubectl get namespace "$namespace" >/dev/null 2>&1; then
-  actual_managed="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.app\\.kubernetes\\.io/managed-by}' 2>/dev/null || true)"
-  actual_target="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/target}' 2>/dev/null || true)"
-  actual_instance="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/instance}' 2>/dev/null || true)"
+current_phase=ownership-check
+write_status running "$current_phase" start-running null
+check_canceled
+if kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s >/dev/null 2>&1; then
+  actual_managed="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.app\\.kubernetes\\.io/managed-by}' 2>/dev/null || true)"
+  actual_target="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/target}' 2>/dev/null || true)"
+  actual_instance="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/instance}' 2>/dev/null || true)"
   if [ "$actual_managed" != "$managed_by" ] || [ "$actual_target" != "$target_id" ] || [ "$actual_instance" != "$instance_id" ]; then
-    printf '%s\n' '{"ok":false,"mutation":false,"code":"namespace-ownership-mismatch","valuesPrinted":false}'
+    failure_code=namespace-ownership-mismatch
     exit 42
   fi
 fi
 printf '%s' ${shQuote(foundationEncoded)} | base64 -d >"$tmp/foundation.yaml"
 printf '%s' ${shQuote(migrationEncoded)} | base64 -d >"$tmp/migration.yaml"
 printf '%s' ${shQuote(workloadsEncoded)} | base64 -d >"$tmp/workloads.yaml"
+current_phase=foundation-apply
+write_status running "$current_phase" start-running null
+check_canceled
 if ! kubectl apply --server-side --force-conflicts --field-manager=${shQuote(target.fieldManager)} -f "$tmp/foundation.yaml" >"$tmp/apply.out" 2>"$tmp/apply.err"; then
-  cat "$tmp/apply.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"apply-failed","valuesPrinted":false}'
+  failure_code=apply-failed
   exit 43
 fi
-wait_rc=0
-kubectl -n "$namespace" rollout status statefulset/pikaoa-postgres --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>"$tmp/wait.err" || wait_rc=$?
-if [ "$wait_rc" -ne 0 ]; then
-  cat "$tmp/wait.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"postgres-wait-failed","valuesPrinted":false}'
+current_phase=postgres-rollout
+write_status running "$current_phase" start-running null
+check_canceled
+if ! kubectl -n "$namespace" rollout status statefulset/pikaoa-postgres --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>"$tmp/wait.err"; then
+  failure_code=postgres-wait-failed
   exit 44
 fi
+current_phase=migration-apply
+write_status running "$current_phase" start-running null
+check_canceled
 if ! kubectl apply --server-side --force-conflicts --field-manager=${shQuote(target.fieldManager)} -f "$tmp/migration.yaml" >"$tmp/migration-apply.out" 2>"$tmp/migration-apply.err"; then
-  cat "$tmp/migration-apply.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"migration-apply-failed","valuesPrinted":false}'
+  failure_code=migration-apply-failed
   exit 45
 fi
+current_phase=migration-wait
+write_status running "$current_phase" start-running null
+check_canceled
 if ! kubectl -n "$namespace" wait --for=condition=complete job/${target.migration.jobName} --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>"$tmp/migration-wait.err"; then
-  cat "$tmp/migration-wait.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"migration-wait-failed","valuesPrinted":false}'
+  failure_code=migration-wait-failed
   exit 46
 fi
+current_phase=workloads-apply
+write_status running "$current_phase" start-running null
+check_canceled
 if ! kubectl apply --server-side --force-conflicts --field-manager=${shQuote(target.fieldManager)} -f "$tmp/workloads.yaml" >"$tmp/workloads-apply.out" 2>"$tmp/workloads-apply.err"; then
-  cat "$tmp/workloads-apply.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"workloads-apply-failed","valuesPrinted":false}'
+  failure_code=workloads-apply-failed
   exit 47
 fi
-kubectl -n "$namespace" rollout status deployment/pikaoa-api --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>"$tmp/wait.err" || wait_rc=$?
-if [ "$wait_rc" -eq 0 ]; then kubectl -n "$namespace" rollout status deployment/pikaoa-worker --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>>"$tmp/wait.err" || wait_rc=$?; fi
-if [ "$wait_rc" -ne 0 ]; then
-  cat "$tmp/wait.err" >&2
-  printf '%s\n' '{"ok":false,"mutation":true,"code":"rollout-wait-failed","valuesPrinted":false}'
+current_phase=api-rollout
+write_status running "$current_phase" start-running null
+check_canceled
+if ! kubectl -n "$namespace" rollout status deployment/pikaoa-api --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>"$tmp/wait.err"; then
+  failure_code=api-rollout-wait-failed
   exit 48
 fi
-printf '%s\n' '{"ok":true,"mutation":true,"code":"started","valuesPrinted":false}'
+current_phase=worker-rollout
+write_status running "$current_phase" start-running null
+check_canceled
+if ! kubectl -n "$namespace" rollout status deployment/pikaoa-worker --timeout=${target.waitTimeoutSeconds}s >/dev/null 2>>"$tmp/wait.err"; then
+  failure_code=worker-rollout-wait-failed
+  exit 49
+fi
 `;
 }
 
 function statusScript(target: TestTargetSpec, context: RenderContext): string {
+  const job = asyncJobIdentity(target, context);
   return `
 set -u
+umask 077
+state_dir=${shQuote(job.stateDir)}
+job_id=${shQuote(job.id)}
 namespace=${shQuote(context.namespace)}
-if ! kubectl get namespace "$namespace" >/dev/null 2>&1; then
-  printf '%s\n' '{"ok":true,"exists":false,"mutation":false,"valuesPrinted":false}'
-  exit 0
+tmp="$(mktemp -d)"
+trap 'rm -rf "$tmp"' EXIT
+runtime_code=runtime-not-found
+if kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o json >"$tmp/namespace.json" 2>"$tmp/namespace.err"; then
+  runtime_code=runtime-present
+  if ! kubectl -n "$namespace" get deployment,statefulset,job,service,persistentvolumeclaim,pod --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o json >"$tmp/resources.json" 2>"$tmp/resources.err"; then
+    runtime_code=runtime-resource-query-failed
+  fi
+elif ! grep -Eiq 'not[[:space:]-]*found' "$tmp/namespace.err"; then
+  runtime_code=runtime-query-failed
 fi
-printf '%s' '{"ok":true,"exists":true,"mutation":false,"namespace":'
-kubectl get namespace "$namespace" -o json
-printf '%s' ',"resources":'
-kubectl -n "$namespace" get deployment,statefulset,job,service,persistentvolumeclaim,pod -o json
-printf '%s\n' ',"valuesPrinted":false}'
+python3 - "$state_dir" "$job_id" "$runtime_code" "$tmp/namespace.json" "$tmp/resources.json" ${shQuote(String(target.taskState.logTailLines))} <<'PY'
+import json, os, pathlib, sys
+state_dir = pathlib.Path(sys.argv[1])
+job_id, runtime_code = sys.argv[2:4]
+namespace_path, resources_path = map(pathlib.Path, sys.argv[4:6])
+tail_lines = int(sys.argv[6])
+status_path = state_dir / "status.json"
+task = None
+task_error = None
+if status_path.exists():
+    try:
+        task = json.loads(status_path.read_text())
+    except Exception:
+        task_error = "start-state-unreadable"
+pid = task.get("pid") if isinstance(task, dict) else None
+runner_alive = False
+if isinstance(pid, int) and pid > 0:
+    try:
+        os.kill(pid, 0)
+        runner_alive = True
+    except OSError:
+        pass
+events = []
+events_path = state_dir / "events.ndjson"
+if events_path.exists():
+    for line in events_path.read_text(errors="replace").splitlines()[-tail_lines:]:
+        try:
+            item = json.loads(line)
+        except Exception:
+            continue
+        if isinstance(item, dict):
+            events.append({key: item.get(key) for key in ("at", "state", "phase", "code", "valuesPrinted")})
+if task_error is not None:
+    code, ok, exists = task_error, False, True
+elif task is None:
+    code, ok, exists = "start-not-found", True, False
+else:
+    state = str(task.get("state") or "unknown")
+    if state in ("queued", "running") and not runner_alive:
+        code, ok = "start-worker-missing", False
+    else:
+        code = {"queued": "start-running", "running": "start-running", "succeeded": "start-succeeded", "failed": "start-failed", "canceled": "start-canceled"}.get(state, "start-state-unknown")
+        ok = state not in ("failed", "canceled", "unknown")
+    exists = True
+task_code = code
+runtime = {"queried": True, "exists": runtime_code in ("runtime-present", "runtime-resource-query-failed"), "code": runtime_code}
+runtime_paths = []
+if runtime_code in ("runtime-present", "runtime-resource-query-failed"):
+    runtime_paths.append(("namespace", namespace_path))
+if runtime_code == "runtime-present":
+    runtime_paths.append(("resources", resources_path))
+for key, path in runtime_paths:
+    if path.exists() and path.stat().st_size > 0:
+        try:
+            runtime[key] = json.loads(path.read_text())
+        except Exception:
+            runtime["code"] = "runtime-response-unreadable"
+    else:
+        runtime["code"] = "runtime-response-unreadable"
+if runtime["code"] in ("runtime-query-failed", "runtime-resource-query-failed", "runtime-response-unreadable"):
+    code, ok = runtime["code"], False
+payload = {"ok": ok, "mutation": False, "code": code, "taskCode": task_code, "jobId": job_id, "taskExists": exists, "task": task, "runnerAlive": runner_alive, "logTail": events, "runtime": runtime, "valuesPrinted": False}
+print(json.dumps(payload, separators=(",", ":")))
+PY
 `;
 }
 
 function stopScript(target: TestTargetSpec, context: RenderContext): string {
+  const job = asyncJobIdentity(target, context);
   return `
 set -u
+umask 077
+state_dir=${shQuote(job.stateDir)}
+job_id=${shQuote(job.id)}
 namespace=${shQuote(context.namespace)}
 managed_by=${shQuote(MANAGED_BY)}
 target_id=${shQuote(target.id)}
 instance_id=${shQuote(context.instanceId)}
-if ! kubectl get namespace "$namespace" >/dev/null 2>&1; then
-  printf '%s\n' '{"ok":true,"mutation":false,"code":"already-absent","valuesPrinted":false}'
+mark_cancel() {
+  python3 - "$state_dir" <<'PY'
+import json, pathlib, sys
+state_dir = pathlib.Path(sys.argv[1])
+status_path = state_dir / "status.json"
+if not status_path.exists():
+    print("not-found|false")
+    raise SystemExit(0)
+try:
+    state = str(json.loads(status_path.read_text()).get("state") or "unknown")
+except Exception:
+    print("state-unreadable|false")
+    raise SystemExit(0)
+if state in ("queued", "running"):
+    cancel_path = state_dir / "cancel-requested"
+    if cancel_path.exists():
+        print("cancel-already-requested|false")
+    else:
+        cancel_path.touch(mode=0o600, exist_ok=False)
+        print("cancel-requested|true")
+else:
+    print("already-terminal|false")
+PY
+}
+tmp="$(mktemp -d)"
+trap 'rm -rf "$tmp"' EXIT
+if ! kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o json >"$tmp/namespace.json" 2>"$tmp/namespace.err"; then
+  if ! grep -Eiq 'not[[:space:]-]*found' "$tmp/namespace.err"; then
+    printf '{"ok":false,"mutation":false,"code":"runtime-query-failed","jobId":"%s","startTaskDisposition":"unknown","valuesPrinted":false}\n' "$job_id"
+    exit 45
+  fi
+  cancel_result="$(mark_cancel)"
+  cancel_disposition="\${cancel_result%%|*}"
+  cancel_mutation="\${cancel_result##*|}"
+  printf '{"ok":true,"mutation":%s,"code":"already-absent","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false}\n' "$cancel_mutation" "$job_id" "$cancel_disposition"
   exit 0
 fi
-actual_managed="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.app\\.kubernetes\\.io/managed-by}' 2>/dev/null || true)"
-actual_target="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/target}' 2>/dev/null || true)"
-actual_instance="$(kubectl get namespace "$namespace" -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/instance}' 2>/dev/null || true)"
+actual_managed="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.app\\.kubernetes\\.io/managed-by}' 2>/dev/null || true)"
+actual_target="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/target}' 2>/dev/null || true)"
+actual_instance="$(kubectl get namespace "$namespace" --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s -o jsonpath='{.metadata.labels.pikaoa\\.unidesk\\.io/instance}' 2>/dev/null || true)"
 if [ "$actual_managed" != "$managed_by" ] || [ "$actual_target" != "$target_id" ] || [ "$actual_instance" != "$instance_id" ]; then
-  printf '%s\n' '{"ok":false,"mutation":false,"code":"namespace-ownership-mismatch","valuesPrinted":false}'
+  printf '{"ok":false,"mutation":false,"code":"namespace-ownership-mismatch","jobId":"%s","startTaskDisposition":"not-touched","valuesPrinted":false}\n' "$job_id"
   exit 42
 fi
-kubectl delete namespace "$namespace" --wait=false >/dev/null
-printf '%s\n' '{"ok":true,"mutation":true,"code":"stop-requested","valuesPrinted":false}'
+cancel_result="$(mark_cancel)"
+cancel_disposition="\${cancel_result%%|*}"
+cancel_mutation="\${cancel_result##*|}"
+if ! kubectl delete namespace "$namespace" --wait=false --request-timeout=${target.taskState.statusRequestTimeoutSeconds}s >/dev/null 2>"$tmp/delete.err"; then
+  printf '{"ok":false,"mutation":%s,"code":"namespace-delete-failed","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false}\n' "$cancel_mutation" "$job_id" "$cancel_disposition"
+  exit 43
+fi
+printf '{"ok":true,"mutation":true,"code":"stop-requested","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false}\n' "$job_id" "$cancel_disposition"
 `;
 }
 
 function summarizeRemoteStatus(parsed: Record<string, unknown>, target: TestTargetSpec, context: RenderContext): Record<string, unknown> {
-  if (parsed.exists !== true) return { ok: parsed.ok === true, exists: false, mutation: false, valuesPrinted: false };
-  const namespace = optionalRecord(parsed.namespace, "status.namespace") ?? {};
+  const runtime = optionalRecord(parsed.runtime, "status.runtime") ?? {};
+  const namespace = optionalRecord(runtime.namespace, "status.runtime.namespace") ?? {};
   const metadata = optionalRecord(namespace.metadata, "status.namespace.metadata") ?? {};
   const labels = optionalRecord(metadata.labels, "status.namespace.metadata.labels") ?? {};
-  const resources = optionalRecord(parsed.resources, "status.resources") ?? {};
+  const resources = optionalRecord(runtime.resources, "status.runtime.resources") ?? {};
   const items = Array.isArray(resources.items) ? resources.items.filter(isRecord) : [];
   return {
     ok: parsed.ok === true,
-    exists: true,
+    code: parsed.code ?? "start-status-unknown",
+    taskCode: parsed.taskCode ?? "start-status-unknown",
+    jobId: parsed.jobId ?? asyncJobIdentity(target, context).id,
+    taskExists: parsed.taskExists === true,
+    task: optionalRecord(parsed.task, "status.task"),
+    runnerAlive: parsed.runnerAlive === true,
+    logTail: Array.isArray(parsed.logTail) ? parsed.logTail.filter(isRecord) : [],
     mutation: false,
-    ownership: {
-      valid: labels["app.kubernetes.io/managed-by"] === MANAGED_BY && labels[TARGET_LABEL] === target.id && labels[INSTANCE_LABEL] === context.instanceId,
-      managedBy: labels["app.kubernetes.io/managed-by"] ?? null,
-      target: labels[TARGET_LABEL] ?? null,
-      instance: labels[INSTANCE_LABEL] ?? null,
+    runtime: {
+      queried: runtime.queried === true,
+      exists: runtime.exists === true,
+      code: runtime.code ?? "runtime-status-unknown",
+      ownership: runtime.exists !== true ? null : {
+        valid: labels["app.kubernetes.io/managed-by"] === MANAGED_BY && labels[TARGET_LABEL] === target.id && labels[INSTANCE_LABEL] === context.instanceId,
+        managedBy: labels["app.kubernetes.io/managed-by"] ?? null,
+        target: labels[TARGET_LABEL] ?? null,
+        instance: labels[INSTANCE_LABEL] ?? null,
+      },
+      resources: items.map((item) => {
+        const itemMetadata = optionalRecord(item.metadata, "resource.metadata") ?? {};
+        const status = optionalRecord(item.status, "resource.status") ?? {};
+        const spec = optionalRecord(item.spec, "resource.spec") ?? {};
+        return {
+          kind: item.kind ?? null,
+          name: itemMetadata.name ?? null,
+          desired: spec.replicas ?? null,
+          ready: status.readyReplicas ?? status.numberReady ?? null,
+          phase: status.phase ?? null,
+        };
+      }),
     },
-    resources: items.map((item) => {
-      const itemMetadata = optionalRecord(item.metadata, "resource.metadata") ?? {};
-      const status = optionalRecord(item.status, "resource.status") ?? {};
-      const spec = optionalRecord(item.spec, "resource.spec") ?? {};
-      return {
-        kind: item.kind ?? null,
-        name: itemMetadata.name ?? null,
-        desired: spec.replicas ?? null,
-        ready: status.readyReplicas ?? status.numberReady ?? null,
-        phase: status.phase ?? null,
-      };
-    }),
     valuesPrinted: false,
   };
 }
@@ -926,12 +1275,14 @@ function renderText(payload: Record<string, unknown>): string {
   const warnings = Array.isArray(payload.warnings) ? payload.warnings.filter(isRecord) : [];
   const blockers = Array.isArray(payload.blockers) ? payload.blockers.filter(isRecord) : [];
   const status = optionalRecord(payload.status, "status");
+  const task = status === null ? null : optionalRecord(status.task, "status.task");
   return [
     `PIKAOA TEST TARGET ${String(payload.action ?? "").split(" ").at(-1)?.toUpperCase() ?? ""} (${payload.ok === false ? "blocked" : "ok"})`,
     `configured=${String(payload.configured === true)} mutation=${String(payload.mutation === true)} config=${String(payload.configPath ?? "-")}`,
     `target=${String(target?.id ?? "-")} node=${String(target?.node ?? "-")} validationOnly=${String(target?.validationOnly ?? "-")}`,
     `instance=${String(instance?.id ?? "-")} namespace=${String(instance?.namespace ?? "-")}`,
-    ...(status === null ? [] : [`status=${status.remoteQueried === false ? String(status.reason ?? "local") : status.exists === false ? "absent" : status.ok === false ? "failed" : "present"}`]),
+    ...(payload.code === undefined ? [] : [`code=${String(payload.code)} jobId=${String(payload.jobId ?? "-")}`]),
+    ...(status === null ? [] : [`status=${status.remoteQueried === false ? String(status.reason ?? "local") : String(status.code ?? "unknown")} phase=${String(task?.phase ?? "-")}`]),
     `warnings=${warnings.length} blockers=${blockers.length} valuesPrinted=false`,
     ...warnings.map((item) => `WARNING ${String(item.code ?? "warning")} blocking=false ${String(item.message ?? "")}`),
     ...blockers.map((item) => `BLOCKER ${String(item.code ?? "blocked")} field=${String(item.field ?? "-")} ${String(item.message ?? "")}`),
@@ -950,6 +1301,7 @@ function targetSummary(target: TestTargetSpec): Record<string, unknown> {
     validationOnly: target.validationOnly,
     namespacePrefix: target.namespacePrefix,
     ttlSeconds: target.ttlSeconds,
+    taskState: target.taskState,
     cleanup: target.cleanup,
   };
 }
@@ -1172,6 +1524,12 @@ function absolutePath(value: unknown, path: string): string {
   return result;
 }
 
+function taskStateRoot(value: unknown, path: string): string {
+  const result = absolutePath(value, path).replace(/\/+$/u, "");
+  if (result.length === 0) throw new Error(`${path} 不能是文件系统根目录`);
+  return result;
+}
+
 function nonEmptyStringList(value: unknown, path: string): string[] {
   if (!Array.isArray(value) || value.length === 0) throw new Error(`${path} 必须是非空字符串数组`);
   return value.map((item, index) => nonEmpty(item, `${path}[${index}]`));
@@ -1193,6 +1551,11 @@ function positiveInteger(value: unknown, path: string, max: number): number {
   return Number(value);
 }
 
+function positiveSafeInteger(value: unknown, path: string): number {
+  if (!Number.isSafeInteger(value) || Number(value) <= 0) throw new Error(`${path} 必须是正安全整数`);
+  return Number(value);
+}
+
 function nonNegativeInteger(value: unknown, path: string): number {
   if (!Number.isSafeInteger(value) || Number(value) < 0) throw new Error(`${path} 必须是非负安全整数`);
   return Number(value);