fix(pikaoa): 允许终态测试步骤安全重跑

docs/MDTODO/details/pikaoa-enterprise-platform/R1.5.9.5_Task_Report.md

@@ -2,7 +2,18 @@
 
 ## 结论
 
-已在 PR #2084 完成 NC01 固定 `pikaoa-test` 测试运行面的 YAML-first Secret 初始化闭环和单步 CLI 收口。测试运行面使用 PK01 host PostgreSQL 独立 `pikaoa_test` role/database、固定 NodePort `32080`，不创建专用测试集群、动态 namespace 或 namespace-local PostgreSQL，也未恢复 dedicated 门禁。
+已在 PR #2084 完成 NC01 固定 `pikaoa-test` 测试运行面的：
+
+- YAML-first Secret 初始化闭环；
+- 单步 CLI；
+- 可重跑异步状态机。
+
+测试运行面边界：
+
+- 使用 PK01 host PostgreSQL 独立 `pikaoa_test` role/database；
+- 固定使用 NodePort `32080`；
+- 不创建专用测试集群、动态 namespace 或 namespace-local PostgreSQL；
+- 不恢复 dedicated 门禁。
 
 ## 实现
 
@@ -18,16 +29,31 @@
   - 新增 `--step foundation`；
   - foundation 不要求 `--commit` 或业务 Secret，只提交带所有权标签的固定 Namespace/PVC；
   - migration、api、worker、web 和 all 继续要求 commit 与 Secret source；
-  - plan 只投影当前 step 所需对象，并保持 Secret 脱敏。
+  - plan 只投影当前 step 所需对象，并保持 Secret 脱敏；
+  - request identity 包含 target、instance、commit、step 和当前 step 稳定 structural manifest fingerprint；
+  - request fingerprint 继续脱敏 Secret `stringData`，并排除每次渲染变化的 `pikaoa.unidesk.io/expires-at`；
+  - 完全相同 request 返回已有任务；
+  - 不同 request 遇 queued/running 且 worker 存活时返回 `start-instance-conflict`；
+  - 不同 request 遇终态或 worker-missing 时，原子退役旧有界状态目录并提交新任务；
+  - `status` 继续读取同一 `target/instance/step` 的最新任务，不引入第二状态库、控制器或租约。
 - `docs/reference/pikaoa.md`：
-  - 删除专用测试集群、动态 namespace 和临时 PostgreSQL 旧设计；
+  - 删除动态 namespace 和临时 PostgreSQL 旧设计；
   - 写明 foundation、Secret sync、migration、api、worker、web 顺序；
+  - 写明终态可重跑、运行中去重和稳定 request fingerprint；
   - 写明先稳定 CLI，再建设独立 test CI/CD，且不得成为业务阻塞门禁。
 
 ## 验证
 
 - `bun --check scripts/src/pikaoa-test-target.ts`：通过。
 - `bun test scripts/src/pikaoa-test-target-async.test.ts`：2 pass，0 fail。
+- 异步回归覆盖：
+  - 同 request 在任务终态后仍返回 `start-already-succeeded`；
+  - 不同 commit 在旧任务运行中返回 `start-instance-conflict`；
+  - 不同 commit 在旧任务终态后重置并重新提交；
+  - 同 commit 但探针 manifest 变化时重置并重新提交；
+  - 仅 `expires-at` 渲染时间变化时仍识别为同 request；
+  - worker-missing 时允许安全重置并重新提交；
+  - foundation 终态后可用新 request 重跑。
 - `bun scripts/cli.ts check --syntax-only`：11 pass，0 fail。
 - `pikaoa test-target plan --step foundation`：
   - `readyForMutation=true`；

docs/reference/pikaoa.md

@@ -89,7 +89,25 @@
 - foundation 不需要 `--commit`，也不检查业务 Secret。
 - migration、api、worker、web 和 all 必须显式提供 `--commit`。
 - 每个 `start` 立即返回任务 ID；使用同一 target、instance 和 step 的 `status` 查询终态。
-- 相同任务重复提交只返回已有任务状态，不并行创建第二任务。
+- request identity 包含：
+  - target；
+  - instance；
+  - 产品 commit；
+  - step；
+  - 当前 step 的稳定 structural manifest fingerprint。
+- request fingerprint 对 Secret `stringData` 只保留 key 结构和脱敏占位，不包含 Secret 原值。
+- request fingerprint 排除 `pikaoa.unidesk.io/expires-at` 等每次渲染变化的易变字段。
+- 完全相同 request 重复提交只返回已有任务状态，不并行创建第二任务。
+- 不同 request 遇旧任务处于 queued/running 且 worker 存活时返回 `start-instance-conflict`。
+- 不同 request 遇旧任务已 succeeded/failed/canceled，或 queued/running 但 worker 明确缺失时：
+  - 原子退役该 step 的旧有界状态目录；
+  - 继续使用同一 `target/instance/step` 状态路径提交新任务；
+  - `status` 只读取该 step 的最新任务。
+- foundation 也遵循相同规则。
+- foundation 需要显式形成新 request 时，可以：
+  - 提供新的 `--commit` 修订标识；
+  - 修改 foundation 的稳定 manifest 结构。
+- 不为重跑引入第二状态库、控制器、租约或长期历史存储。
 - 迁移、rollout 或运行面查询失败返回具名失败码，不静默进入下一阶段。
 - OTel exporter、配置版本和 commit 对齐漂移只产生 `blocking=false` warning，不恢复 dedicated 门禁。
 

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

@@ -149,6 +149,12 @@ exit 64
     const result = await runPikaoaCommand(config, args(action, instance, withCommit, step), localCapture);
     return JSON.parse(result.renderedText) as Record<string, unknown>;
   };
+  const projectionWithCommit = async (action: string, instance: string, commit: string, step = "all"): Promise<Record<string, unknown>> => {
+    const commandArgs = args(action, instance, true, step);
+    commandArgs[commandArgs.indexOf("--commit") + 1] = commit;
+    const result = await runPikaoaCommand(config, commandArgs, localCapture);
+    return JSON.parse(result.renderedText) as Record<string, unknown>;
+  };
   const waitFor = async (instance: string, expectedCode: string, step = "all"): Promise<Record<string, unknown>> => {
     for (let attempt = 0; attempt < 80; attempt += 1) {
       const value = await projection("status", instance, false, step);
@@ -186,6 +192,11 @@ exit 64
     });
     const running = await projection("status", "runtime");
     assert.equal((running.status as Record<string, unknown>).code, "start-running");
+    const concurrentCommit = "fedcba9876543210";
+    const concurrent = await projectionWithCommit("start", "runtime", concurrentCommit);
+    assert.equal(concurrent.code, "start-instance-conflict");
+    assert.equal(concurrent.mutation, false);
+    assert.equal(concurrent.previousTaskReset, false);
     rmSync(join(fakeRoot, "slow"));
     const succeeded = await waitFor("runtime", "start-succeeded");
     const succeededStatus = succeeded.status as Record<string, unknown>;
@@ -215,6 +226,32 @@ exit 64
     const repeated = await projection("start", "runtime", true);
     assert.equal(repeated.code, "start-already-succeeded");
     assert.equal(repeated.mutation, false);
+    const rerunSubmitted = await projectionWithCommit("start", "runtime", concurrentCommit);
+    assert.equal(rerunSubmitted.code, "start-submitted");
+    assert.equal(rerunSubmitted.previousTaskReset, true);
+    const rerunSucceeded = await waitFor("runtime", "start-succeeded");
+    assert.equal(((rerunSucceeded.status as Record<string, unknown>).task as Record<string, unknown>).sourceCommit, concurrentCommit);
+    writeFileSync(fixturePath, readFileSync(fixturePath, "utf8").replace("          livenessPath: /healthz\n", "          livenessPath: /healthz-v2\n"));
+    const manifestRerunSubmitted = await projectionWithCommit("start", "runtime", concurrentCommit);
+    assert.equal(manifestRerunSubmitted.code, "start-submitted");
+    assert.equal(manifestRerunSubmitted.previousTaskReset, true);
+    const manifestRerunSucceeded = await waitFor("runtime", "start-succeeded");
+    await Bun.sleep(5);
+    const sameRequestAfterRenderTimeChanged = await projectionWithCommit("start", "runtime", concurrentCommit);
+    assert.equal(sameRequestAfterRenderTimeChanged.code, "start-already-succeeded");
+    assert.equal(sameRequestAfterRenderTimeChanged.previousTaskReset, false);
+
+    writeFileSync(join(fakeRoot, "slow"), "1\n");
+    const workerMissingSubmitted = await projection("start", "runtime", true, "api");
+    const workerPid = Number((workerMissingSubmitted.remote as Record<string, unknown>).pid);
+    assert.equal(Number.isSafeInteger(workerPid) && workerPid > 0, true);
+    process.kill(workerPid, "SIGKILL");
+    await Bun.sleep(25);
+    const workerMissingReset = await projectionWithCommit("start", "runtime", concurrentCommit, "api");
+    assert.equal(workerMissingReset.code, "start-submitted");
+    assert.equal(workerMissingReset.previousTaskReset, true);
+    rmSync(join(fakeRoot, "slow"));
+    const workerMissingSucceeded = await waitFor("runtime", "start-succeeded", "api");
 
     writeFileSync(join(fakeRoot, "fail-migration"), "1\n");
     await projection("start", "runtime", true, "migration");
@@ -246,14 +283,39 @@ exit 64
     const foundationSubmitted = await projection("start", "foundation", false, "foundation");
     assert.equal(foundationSubmitted.code, "start-submitted", JSON.stringify(lastCapture));
     const foundationSucceeded = await waitFor("foundation", "start-succeeded", "foundation");
+    const foundationRepeated = await projection("start", "foundation", false, "foundation");
+    assert.equal(foundationRepeated.code, "start-already-succeeded");
+    const foundationRerun = await projectionWithCommit("start", "foundation", concurrentCommit, "foundation");
+    assert.equal(foundationRerun.code, "start-submitted");
+    assert.equal(foundationRerun.previousTaskReset, true);
+    const foundationRerunSucceeded = await waitFor("foundation", "start-succeeded", "foundation");
 
-    const serialized = JSON.stringify({ submitted, succeeded, failed, stopped, foundationSubmitted, foundationSucceeded });
+    const serialized = JSON.stringify({
+      submitted,
+      succeeded,
+      concurrent,
+      rerunSubmitted,
+      rerunSucceeded,
+      manifestRerunSubmitted,
+      manifestRerunSucceeded,
+      sameRequestAfterRenderTimeChanged,
+      workerMissingSubmitted,
+      workerMissingReset,
+      workerMissingSucceeded,
+      failed,
+      stopped,
+      foundationSubmitted,
+      foundationSucceeded,
+      foundationRepeated,
+      foundationRerun,
+      foundationRerunSucceeded,
+    });
     for (const value of secretValues) assert.equal(serialized.includes(value), false);
     assert.equal(serialized.includes("postgresql://pikaoa_test:"), false);
   } finally {
     rmSync(root, { recursive: true, force: true });
   }
-}, 15_000);
+}, 30_000);
 
 test("validation-only fixture never invokes remote capture", async () => {
   let calls = 0;

scripts/src/pikaoa-test-target.ts

@@ -188,7 +188,7 @@ export function pikaoaTestTargetHelp(): Record<string, unknown> {
     ],
     source: "config/pikaoa.yaml#testRuntime",
     guarantees: [
-      "未声明专用 target 时 plan/status 返回 configured=false，start/stop 在远端调用前失败。",
+      "未声明测试 target 时 plan/status 返回 configured=false，start/stop 在远端调用前失败。",
       "validationOnly target 只用于本地渲染，不连接 route。",
       "start 可按 foundation、migration、api、worker、web 或 all 单步提交并立即返回；foundation 只创建固定 Namespace/PVC，不要求业务 Secret 已存在。",
       "正式 delivery namespace、PaC、GitOps、公网暴露和生产数据库不属于该入口的写入范围。",
@@ -589,6 +589,7 @@ async function startResult(config: UniDeskConfig, options: TestTargetOptions, se
     mode: mutation ? "submitted" : parsed === null ? "submit-failed" : "existing-task",
     code: parsed?.code ?? "start-submit-unreadable",
     jobId: parsed?.jobId ?? asyncJobIdentity(selection.target, context).id,
+    previousTaskReset: parsed?.previousTaskReset === true,
     task: parsed?.task ?? null,
     secret: { sources: material.sources, valuesPrinted: false },
     remote: parsed ?? compactCapture(remote, { full: true }),
@@ -648,7 +649,7 @@ function safetyBlockers(options: TestTargetOptions, selection: Selection, contex
   const blockers: Blocker[] = [];
   const target = selection.target;
   if (target === null) {
-    blockers.push({ code: "test-target-not-configured", field: `${selection.configLabel}#testRuntime.targets`, message: "没有选中已启用的专用测试 target。" });
+    blockers.push({ code: "test-target-not-configured", field: `${selection.configLabel}#testRuntime.targets`, message: "没有选中已启用的测试 target。" });
     return blockers;
   }
   if (requireCommit && options.commit === null) blockers.push({ code: "commit-required", field: "--commit", message: "当前 step 必须选择源码 commit。" });
@@ -938,6 +939,7 @@ function envValue(raw: string, key: string): string {
 
 function startScript(target: TestTargetSpec, context: RenderContext, manifest: Record<string, unknown>[]): string {
   const job = asyncJobIdentity(target, context);
+  const manifestFingerprint = requestStructuralFingerprint(manifest);
   const requestId = createHash("sha256")
     .update(target.id)
     .update("\0")
@@ -946,6 +948,8 @@ function startScript(target: TestTargetSpec, context: RenderContext, manifest: R
     .update(context.commit)
     .update("\0")
     .update(context.step)
+    .update("\0")
+    .update(manifestFingerprint)
     .digest("hex");
   const runnerEncoded = Buffer.from(startRunnerScript(target, context, manifest, job), "utf8").toString("base64");
   return `
@@ -958,29 +962,38 @@ request_file="$state_dir/request-id"
 pid_file="$state_dir/pid"
 job_id=${shQuote(job.id)}
 request_id=${shQuote(requestId)}
-emit_existing() {
+reset_previous=false
+inspect_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)
+    raise SystemExit(10)
 state = str(task.get("state") or "unknown")
 runner_alive = False
 pid = task.get("pid")
+if not isinstance(pid, int) or pid <= 0:
+    try:
+        pid = int(pid_path.read_text().strip())
+    except Exception:
+        pid = None
 if isinstance(pid, int) and pid > 0:
     try:
         os.kill(pid, 0)
         runner_alive = True
     except OSError:
         pass
+worker_missing = state in ("queued", "running") and not runner_alive
+if actual_request != expected_request:
+    if state in ("succeeded", "failed", "canceled") or worker_missing:
+        raise SystemExit(20)
+    print(json.dumps({"ok": False, "mutation": False, "code": "start-instance-conflict", "jobId": job_id, "submitted": False, "task": task, "valuesPrinted": False}))
+    raise SystemExit(10)
 code_by_state = {
     "queued": "start-already-running",
     "running": "start-already-running",
@@ -988,20 +1001,40 @@ code_by_state = {
     "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")
+code = "start-worker-missing" if worker_missing 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}))
+raise SystemExit(10)
 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
+claim_attempt=0
+while ! mkdir "$state_dir" 2>/dev/null; do
+  if [ ! -d "$state_dir" ]; then
+    printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-create-failed","valuesPrinted":false}'
+    exit 42
+  fi
+  inspect_existing
+  decision=$?
+  if [ "$decision" -eq 10 ]; then exit 0; fi
+  if [ "$decision" -ne 20 ]; then
+    printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-inspect-failed","valuesPrinted":false}'
+    exit 42
+  fi
+  retired_dir="$state_dir.reset.$request_id.$$"
+  if mv "$state_dir" "$retired_dir" 2>/dev/null; then
+    rm -rf "$retired_dir"
+    reset_previous=true
+  fi
+  claim_attempt=$((claim_attempt + 1))
+  if [ "$claim_attempt" -ge 5 ]; then
+    printf '%s\n' '{"ok":false,"mutation":false,"code":"task-state-reset-conflict","valuesPrinted":false}'
+    exit 42
+  fi
+done
 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"
@@ -1045,7 +1078,7 @@ if payload.get("pid") is None and payload.get("phase") == "submitted":
     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"
+printf '{"ok":true,"mutation":true,"code":"start-submitted","submitted":true,"previousTaskReset":%s,"jobId":"%s","pid":%s,"task":{"state":"running","phase":"submitted","code":"start-running"},"valuesPrinted":false}\n' "$reset_previous" "$job_id" "$pid"
 `;
 }
 
@@ -1513,8 +1546,28 @@ function objectSummary(object: Record<string, unknown>): Record<string, unknown>
 }
 
 function structuralFingerprint(manifest: Record<string, unknown>[]): string {
-  const redacted = manifest.map((object) => object.kind === "Secret" ? { ...object, stringData: Object.fromEntries(Object.keys(record(object.stringData, "Secret.stringData")).sort().map((key) => [key, "<redacted>"])) } : object);
-  return `sha256:${createHash("sha256").update(JSON.stringify(redacted)).digest("hex")}`;
+  return manifestFingerprint(manifest, false);
+}
+
+function requestStructuralFingerprint(manifest: Record<string, unknown>[]): string {
+  return manifestFingerprint(manifest, true);
+}
+
+function manifestFingerprint(manifest: Record<string, unknown>[], stableRequest: boolean): string {
+  const normalized = manifest.map((object) => {
+    let value = object.kind === "Secret"
+      ? { ...object, stringData: Object.fromEntries(Object.keys(record(object.stringData, "Secret.stringData")).sort().map((key) => [key, "<redacted>"])) }
+      : object;
+    if (!stableRequest) return value;
+    const metadata = optionalRecord(value.metadata, "metadata");
+    const annotations = metadata === null ? null : optionalRecord(metadata.annotations, "metadata.annotations");
+    if (metadata === null || annotations === null || annotations["pikaoa.unidesk.io/expires-at"] === undefined) return value;
+    const stableAnnotations = { ...annotations };
+    delete stableAnnotations["pikaoa.unidesk.io/expires-at"];
+    value = { ...value, metadata: { ...metadata, annotations: stableAnnotations } };
+    return value;
+  });
+  return `sha256:${createHash("sha256").update(JSON.stringify(normalized)).digest("hex")}`;
 }
 
 function placeholderSecrets(): SecretMaterial {