diff --git a/scripts/src/cli.ts b/scripts/src/cli.ts index 6c96b34..36884bf 100644 --- a/scripts/src/cli.ts +++ b/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 { 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 { + 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 { const aipod = optionalFlag(args, "aipod") ?? optionalFlag(args, "aipod-spec"); if (aipod) return submitQueueTaskWithAipod(args, aipod); diff --git a/src/selftest/cases/76-aipod-spec.ts b/src/selftest/cases/76-aipod-spec.ts index 0ae7b24..e4b2a26 100644 --- a/src/selftest/cases/76-aipod-spec.ts +++ b/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 ")), 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((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 { + 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((resolve) => proc.on("close", resolve))]); + assert.notEqual(code, 0, stderr || stdout); + return JSON.parse(stdout) as JsonRecord; +} + async function readStream(stream: NodeJS.ReadableStream): Promise { const chunks: Buffer[] = []; stream.on("data", (chunk: Buffer | string) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));