Merge pull request #351 from pikasTech/fix/r6-5-session-durable-identity

fix(cli): 取消态 Aipod 会话继承持久身份

scripts/src/cli.ts

@@ -1570,7 +1570,13 @@ async function submitQueueTaskWithAipod(args: ParsedArgs, aipod: string): Promis
 
 async function sessionSendWithAipod(args: ParsedArgs, positionalSessionId: string | null, aipod: string): Promise<JsonRecord> {
   const sessionId = positionalSessionId ?? optionalFlag(args, "session-id") ?? newSessionId();
-  const rendered = await renderAipodForCommand(args, aipod, positionalSessionId ? 3 : 2, { sessionId });
+  const existing = await fetchDurableSessionForAipodSend(args, sessionId);
+  const renderInput = inheritDurableSessionAipodIdentity(
+    await aipodRenderInput(args, positionalSessionId ? 3 : 2, { sessionId }),
+    sessionId,
+    existing,
+  );
+  const rendered = await client(args).post(`/api/v1/aipod-specs/${encodeURIComponent(aipod)}/render`, renderInput) as RenderedAipodQueueTask;
   const task = rendered.queueTask;
   const profile = String(task.backendProfile);
   const sessionRef = objectField(task as unknown as JsonRecord, "sessionRef", {});
@@ -1605,6 +1611,70 @@ async function sessionSendWithAipod(args: ParsedArgs, positionalSessionId: strin
   return { ...summarizeSessionSendResult(result, sessionId, profile, aipod), modelResolution: rendered.modelResolution };
 }
 
+async function fetchDurableSessionForAipodSend(args: ParsedArgs, sessionId: string): Promise<JsonRecord | null> {
+  try {
+    return await client(args).get(`/api/v1/sessions/${encodeURIComponent(sessionId)}`) as JsonRecord;
+  } catch (error) {
+    if (error instanceof AgentRunError && error.httpStatus === 404) return null;
+    throw error;
+  }
+}
+
+export function inheritDurableSessionAipodIdentity(
+  renderInput: RenderAipodInput,
+  sessionId: string,
+  existing: JsonRecord | null,
+): RenderAipodInput {
+  if (existing === null) return renderInput;
+  const durableSessionId = stringValue(existing.sessionId);
+  const tenantId = stringValue(existing.tenantId);
+  const projectId = stringValue(existing.projectId);
+  if (durableSessionId !== sessionId || tenantId === null || projectId === null) {
+    throw new AgentRunError("schema-invalid", `session ${sessionId} durable identity is incomplete or mismatched`, {
+      httpStatus: 400,
+      details: { sessionId, valuesPrinted: false },
+    });
+  }
+  assertDurableSessionBoundary(renderInput.tenantId, tenantId, sessionId, "tenantId");
+  assertDurableSessionBoundary(renderInput.projectId, projectId, sessionId, "projectId");
+  const requestedSessionRef = jsonRecordValue(renderInput.sessionRef) ?? {};
+  const requestedSessionId = stringValue(requestedSessionRef.sessionId);
+  if (requestedSessionId !== null && requestedSessionId !== sessionId) {
+    throw durableSessionBoundaryError(sessionId, "sessionRef.sessionId");
+  }
+  const conversationId = stringValue(existing.conversationId);
+  const threadId = stringValue(existing.threadId);
+  const expiresAt = stringValue(existing.expiresAt);
+  return {
+    ...renderInput,
+    tenantId,
+    projectId,
+    sessionId,
+    sessionRef: {
+      sessionId,
+      ...(conversationId === null ? {} : { conversationId }),
+      ...(threadId === null ? {} : { threadId }),
+      ...(expiresAt === null ? {} : { expiresAt }),
+      metadata: {
+        ...(jsonRecordValue(requestedSessionRef.metadata) ?? {}),
+        ...(jsonRecordValue(existing.metadata) ?? {}),
+      },
+    },
+  };
+}
+
+function assertDurableSessionBoundary(value: string | undefined, durable: string, sessionId: string, field: string): void {
+  if (value === undefined || value === durable) return;
+  throw durableSessionBoundaryError(sessionId, field);
+}
+
+function durableSessionBoundaryError(sessionId: string, field: string): AgentRunError {
+  return new AgentRunError("tenant-policy-denied", "sessionRef cannot be reused across tenant or project boundary", {
+    httpStatus: 403,
+    details: { sessionId, field, valuesPrinted: false },
+  });
+}
+
 async function submitQueueTask(args: ParsedArgs): Promise<JsonValue> {
   const aipod = optionalFlag(args, "aipod") ?? optionalFlag(args, "aipod-spec");
   if (aipod) return submitQueueTaskWithAipod(args, aipod);

src/selftest/cases/76-aipod-spec.ts

@@ -8,7 +8,7 @@ import { startManagerServer } from "../../mgr/server.js";
 import { MemoryAgentRunStore } from "../../mgr/store.js";
 import type { JsonRecord } from "../../common/types.js";
 import { validateAipodSpec } from "../../common/aipod-specs.js";
-import { validatePrimaryWorkspaceContract, validateResourceBundleRef } from "../../common/validation.js";
+import { validateCreateCommand, validateCreateRun, validatePrimaryWorkspaceContract, validateResourceBundleRef } from "../../common/validation.js";
 import { classifyGitStderr, resolveGitBundleFetchSource } from "../../runner/resource-bundle.js";
 import { assertNoSecretLeak, loadArtificerImageRef, type SelfTestCase } from "../harness.js";
 
@@ -17,12 +17,18 @@ const selfTest: SelfTestCase = async (context) => {
   const artificerSpecFile = path.join(aipodSpecDir, "artificer.yaml");
   await mkdir(aipodSpecDir, { recursive: true });
   await copyFile(path.join(context.root, "config", "aipods", "artificer.yaml"), artificerSpecFile);
+  const store = new MemoryAgentRunStore();
   const server = await startManagerServer({
     port: 0,
     host: "127.0.0.1",
     sourceCommit: "self-test",
-    store: new MemoryAgentRunStore(),
+    store,
     aipodSpecDir,
+    sessionPvcOptions: {
+      kubectlHandler: ({ args }) => args[0] === "get"
+        ? { stdout: JSON.stringify({ status: { phase: "Bound", capacity: { storage: "1Gi" } } }), stderr: "", exitCode: 0 }
+        : { stdout: "{}", stderr: "", exitCode: 0 },
+    },
     runnerJobDefaults: {
       namespace: "agentrun-v02",
       managerUrl: "http://agentrun-mgr.agentrun-v02.svc.cluster.local:8080",
@@ -221,6 +227,52 @@ const selfTest: SelfTestCase = async (context) => {
     assert.equal((sendPlanData.modelResolution as JsonRecord).reasoningEffort, "medium");
     assert.equal((((sendPlanData.modelResolution as JsonRecord).providerCredential as JsonRecord).secretRef as JsonRecord).name, "agentrun-v02-provider-gpt-pika");
 
+    const cancelledSessionId = "sess_artificer_cancelled_cli_selftest";
+    const cancelledConversationId = "conv_artificer_cancelled_cli_selftest";
+    const cancelledThreadId = "thread_artificer_cancelled_cli_selftest";
+    const cancelledRun = store.createRun(validateCreateRun({
+      tenantId: "unidesk",
+      projectId: "pikainc/selfmedia",
+      providerId: task.providerId,
+      backendProfile: task.backendProfile,
+      workspaceRef: task.workspaceRef,
+      sessionRef: {
+        sessionId: cancelledSessionId,
+        conversationId: cancelledConversationId,
+        threadId: cancelledThreadId,
+        expiresAt: "2030-01-01T00:00:00.000Z",
+        metadata: { lastTurnId: "turn_cancelled_cli_selftest", durable: true },
+      },
+      executionPolicy: task.executionPolicy,
+      resourceBundleRef: task.resourceBundleRef,
+      traceSink: { kind: "queue", taskId: "qt_cancelled_cli_selftest", aipod: "Artificer" },
+    }));
+    const cancelledCommand = store.createCommand(cancelledRun.id, validateCreateCommand({ type: "turn", payload: { prompt: "cancelled artificer turn" } }));
+    store.finishCommand(cancelledCommand.id, { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: "self-test cancellation" });
+    store.finishRun(cancelledRun.id, { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: "self-test cancellation" });
+    assert.equal(store.getSession(cancelledSessionId)?.lastRunId, cancelledRun.id);
+
+    const conflict = await runCliFailureJson(context, server.baseUrl, ["sessions", "send", cancelledSessionId, "--aipod", "Artificer", "--project-id", "pikasTech/unidesk", "--prompt", "must remain fail closed", "--no-runner-job"]);
+    assert.equal(conflict.ok, false);
+    assert.match(JSON.stringify(conflict), /tenant-policy-denied/u);
+    assert.equal(store.getSession(cancelledSessionId)?.lastRunId, cancelledRun.id, "boundary conflict must not mutate the cancelled session");
+
+    const continued = await runCliJson(context, server.baseUrl, ["sessions", "send", cancelledSessionId, "--aipod", "Artificer", "--prompt", "continue cancelled selfmedia task", "--no-runner-job"]);
+    assert.equal(continued.ok, true);
+    const continuedData = continued.data as JsonRecord;
+    assert.equal(continuedData.action, "session-send");
+    assert.equal(continuedData.decision, "turn");
+    const continuedSession = store.getSession(cancelledSessionId);
+    assert.notEqual(continuedSession?.lastRunId, cancelledRun.id);
+    const continuedRun = store.getRun(String(continuedSession?.lastRunId));
+    assert.equal(continuedRun.tenantId, "unidesk");
+    assert.equal(continuedRun.projectId, "pikainc/selfmedia");
+    assert.equal(continuedRun.sessionRef?.sessionId, cancelledSessionId);
+    assert.equal(continuedRun.sessionRef?.conversationId, cancelledConversationId);
+    assert.equal(continuedRun.sessionRef?.threadId, cancelledThreadId);
+    assert.equal(continuedRun.sessionRef?.metadata?.lastTurnId, "turn_cancelled_cli_selftest");
+    assert.equal((continuedRun.traceSink as JsonRecord).aipod, "Artificer");
+
     const help = await runCliJson(context, server.baseUrl, ["help"]);
     const commands = ((help.data as JsonRecord).commands as string[]) ?? [];
     assert.equal(commands.some((item) => item.includes("aipod-specs render <name>")), true);
@@ -240,7 +292,7 @@ const selfTest: SelfTestCase = async (context) => {
     assert.equal((storedTask.payload as JsonRecord).model, "gpt-5.6-sol");
     assert.equal((storedTask.payload as JsonRecord).reasoningEffort, "medium");
     assert.deepEqual((storedTask.payload as JsonRecord).modelConfig, taskPayload.modelConfig);
-    return { name: "aipod-spec", tests: ["aipod-spec-yaml-parser-runtime-compatible", "aipod-spec-artificer-image-ref-render", "aipod-spec-artificer-v02-runtime-authority", "aipod-spec-gpt-pika-secret-namespace", "aipod-spec-model-defaults", "aipod-spec-model-reasoning-overrides", "aipod-spec-model-override-keeps-provider", "aipod-spec-control-override-fail-closed", "aipod-spec-payload-control-bypass-fail-closed", "aipod-spec-missing-gpt-pika-credential", "aipod-spec-durable-task-model-snapshot", "aipod-spec-primary-workspace-contract", "aipod-spec-artificer-github-url-render", "aipod-spec-artificer-github-ssh-required-keys", "aipod-spec-git-mirror-url", "git-fetch-stderr-classification", "resource-bundle-source-authority-validation", "resource-bundle-target-conflict-validation", "queue-submit-aipod-dry-run", "session-send-aipod-model-dry-run", "aipod-cli-help"] };
+    return { name: "aipod-spec", tests: ["aipod-spec-yaml-parser-runtime-compatible", "aipod-spec-artificer-image-ref-render", "aipod-spec-artificer-v02-runtime-authority", "aipod-spec-gpt-pika-secret-namespace", "aipod-spec-model-defaults", "aipod-spec-model-reasoning-overrides", "aipod-spec-model-override-keeps-provider", "aipod-spec-control-override-fail-closed", "aipod-spec-payload-control-bypass-fail-closed", "aipod-spec-missing-gpt-pika-credential", "aipod-spec-durable-task-model-snapshot", "aipod-spec-primary-workspace-contract", "aipod-spec-artificer-github-url-render", "aipod-spec-artificer-github-ssh-required-keys", "aipod-spec-git-mirror-url", "git-fetch-stderr-classification", "resource-bundle-source-authority-validation", "resource-bundle-target-conflict-validation", "queue-submit-aipod-dry-run", "session-send-aipod-model-dry-run", "session-send-aipod-cancelled-durable-identity", "session-send-aipod-boundary-conflict-fail-closed", "aipod-cli-help"] };
   } finally {
     await new Promise<void>((resolve) => server.server.close(() => resolve()));
   }
@@ -266,6 +318,13 @@ async function runCliJson(context: { root: string }, managerUrl: string, args: s
   return JSON.parse(stdout) as JsonRecord;
 }
 
+async function runCliFailureJson(context: { root: string }, managerUrl: string, args: string[]): Promise<JsonRecord> {
+  const proc = spawn(process.execPath, [`${context.root}/scripts/agentrun-cli.ts`, "--manager-url", managerUrl, ...args], { stdio: ["ignore", "pipe", "pipe"] });
+  const [stdout, stderr, code] = await Promise.all([readStream(proc.stdout), readStream(proc.stderr), new Promise<number | null>((resolve) => proc.on("close", resolve))]);
+  assert.notEqual(code, 0, stderr || stdout);
+  return JSON.parse(stdout) as JsonRecord;
+}
+
 async function readStream(stream: NodeJS.ReadableStream): Promise<string> {
   const chunks: Buffer[] = [];
   stream.on("data", (chunk: Buffer | string) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));