From e1cb5b699e8b523952f5710a452e04fe30e540df Mon Sep 17 00:00:00 2001 From: AIGameCreator App Date: Thu, 16 Jul 2026 12:38:34 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E9=BD=90=E5=8D=95=20Agent=20=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8F=AA=E8=AF=BB=E5=B9=B6=E8=A1=8C=E6=89=B9=E6=AC=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现严格只读动作的持久并行执行与线性化恢复 补齐崩溃窗口、控制漂移和串行屏障回归 新增真实 Provider parallel-read suite 并记录 PASS 证据 同步 Runtime 技术方案、实施计划与项目决策 --- .../scripts/agent-runtime-real-e2e.mjs | 752 ++++++- .../src-tauri/src/agent.rs | 1853 ++++++++++++++++- .../src-tauri/src/tests.rs | 805 ++++++- .../shared-memory/decision-log.md | 10 + ...案】AI游戏创作Agent Runtime V1.1-2026-07-12.md | 15 + ...案】AI游戏创作智能体App实施计划-2026-06-24.md | 2 + 6 files changed, 3412 insertions(+), 25 deletions(-) diff --git a/apps/ai-game-creator-shell/scripts/agent-runtime-real-e2e.mjs b/apps/ai-game-creator-shell/scripts/agent-runtime-real-e2e.mjs index 0ebbb0cc6..7be682406 100644 --- a/apps/ai-game-creator-shell/scripts/agent-runtime-real-e2e.mjs +++ b/apps/ai-game-creator-shell/scripts/agent-runtime-real-e2e.mjs @@ -51,6 +51,10 @@ const steerRunnerKillAppDataSentinelFileName = '.agent-runtime-real-e2e-steer-runner-kill-appdata.json'; const steerRunnerKillAppDataSentinelSchema = 'genarrative-agent-runtime-real-e2e-steer-runner-kill-appdata.v1'; +const parallelReadAppDataSentinelFileName = + '.agent-runtime-real-e2e-parallel-read-appdata.json'; +const parallelReadAppDataSentinelSchema = + 'genarrative-agent-runtime-real-e2e-parallel-read-appdata.v1'; const mainAgentId = 'code-prototype'; const projectSupervisorAgentId = 'project-supervisor'; const requestedRunId = `real-e2e-${Date.now()}-${randomUUID().slice(0, 8)}`; @@ -78,6 +82,7 @@ const userInputRuntimeSuite = 'user-input-runtime'; const scopedAgentsSuite = 'scoped-agents'; const projectSkillSuite = 'project-skill'; const steerRunnerKillSuite = 'steer-runner-kill'; +const parallelReadSuite = 'parallel-read'; const runtimeContextBundleSchemaVersion = 'game-creator-runtime-context-bundle.v5'; const providerRequestLifecycleSchemaVersion = @@ -127,6 +132,14 @@ const projectSkillIrrelevantEntryPath = '.codex/skills/unrelated-art/SKILL.md'; const projectSkillTargetPath = 'game/release-capsule.txt'; const projectSkillVerificationScriptPath = 'verify-project-skill.mjs'; const projectSkillVerificationCommand = `node ${projectSkillVerificationScriptPath}`; +const parallelReadCorpusPath = 'game/parallel-read-corpus'; +const parallelReadAlphaQuery = `PARALLEL_READ_ALPHA_${randomUUID() + .replaceAll('-', '') + .slice(0, 20)}`; +const parallelReadBetaQuery = `PARALLEL_READ_BETA_${randomUUID() + .replaceAll('-', '') + .slice(0, 20)}`; +const parallelReadCorpusFileCount = 420; const webSearchBaselineApiUrl = 'https://api.github.com/repos/nodejs/node/releases/latest'; const goalSessionId = `agent-session-${mainAgentId}`; @@ -464,6 +477,12 @@ const state = { privateValues: [], reportLeakCount: 0, }, + parallelRead: { + effectiveModel: null, + effectiveApiKind: null, + privateValues: [], + reportLeakCount: 0, + }, confirmedActionIds: new Set(), cleanupPerformed: false, process: { @@ -523,6 +542,7 @@ try { if (isUserInputRuntimeSuite()) state.evidence = emptyUserInputEvidence(); if (isScopedAgentsSuite()) state.evidence = emptyScopedAgentsEvidence(); if (isProjectSkillSuite()) state.evidence = emptyProjectSkillEvidence(); + if (isParallelReadSuite()) state.evidence = emptyParallelReadEvidence(); if (isSteerRunnerKillSuite()) { state.evidence = emptySteerRunnerKillEvidence(); } @@ -534,6 +554,7 @@ try { isUserInputRuntimeSuite() || isScopedAgentsSuite() || isProjectSkillSuite() || + isParallelReadSuite() || isSteerRunnerKillSuite() ) { state.formalConfigPathTranscriptScanner = new StreamingSecretScanner( @@ -575,6 +596,8 @@ try { await runScopedAgentsE2e(); } else if (isProjectSkillSuite()) { await runProjectSkillE2e(); + } else if (isParallelReadSuite()) { + await runParallelReadE2e(); } else if (isProcessSessionSuite()) { await runProcessSessionE2e(); } else { @@ -793,6 +816,28 @@ try { state.status = 'FAIL'; recordError('project-skill-formal-config-cli-call-detected'); } + } else if (isParallelReadSuite()) { + state.evidence.parallelReadRunnerStopped = state.isolatedRunner.stopped; + state.evidence.parallelReadAppDataCleanupPerformed = + state.isolatedRunner.cleanupPerformed; + state.evidence.parallelReadRunnerKillMethod = killMethod; + state.evidence.parallelReadRunnerPidfdClaimCount = + state.isolatedRunner.pidfdClaimCount; + state.evidence.parallelReadRunnerPidfdSignalCount = + state.isolatedRunner.pidfdSignalCount; + state.evidence.formalConfigCliCallCount = + state.isolatedRunner.sourceConfigCliCallCount; + state.evidence.sourceRunnerEndpointUnchanged = + state.isolatedRunner.sourceRunnerEndpointUnchanged; + state.evidence.sourceConfigReplicaCount = + state.isolatedRunner.configLinks.length; + state.evidence.sourceConfigReplicasVerified = + state.isolatedRunner.sourceConfigLinksVerified; + state.evidence.isolatedAppDataUsed = true; + if (state.isolatedRunner.sourceConfigCliCallCount > 0) { + state.status = 'FAIL'; + recordError('parallel-read-formal-config-cli-call-detected'); + } } else { assert( isContextCompactionSuite(), @@ -924,6 +969,16 @@ try { recordError('project-skill-partial-evidence-read-failed', error); } } + if (isParallelReadSuite() && state.projectRoot && state.status !== 'PASS') { + try { + state.evidence = { + ...state.evidence, + ...(await collectPartialParallelReadEvidence()), + }; + } catch (error) { + recordError('parallel-read-partial-evidence-read-failed', error); + } + } if (state.projectRoot && state.secrets.length > 0) { try { state.projectLeakCount = await countSecretsInProject( @@ -1131,6 +1186,20 @@ try { report = JSON.stringify(summary, null, 2); } } + if (isParallelReadSuite()) { + state.parallelRead.reportLeakCount = countExactSecrets( + Buffer.from(report), + state.parallelRead.privateValues, + ); + state.evidence.parallelReadReportLeakCount = + state.parallelRead.reportLeakCount; + if (state.parallelRead.reportLeakCount > 0) { + state.status = 'FAIL'; + recordError('parallel-read-private-body-report-leak-detected'); + summary = buildSummary(); + report = JSON.stringify(summary, null, 2); + } + } state.projectPathReportLeakCount = countExactSecrets( Buffer.from(report), disposableProjectPathVariants(), @@ -1149,6 +1218,7 @@ try { isUserInputRuntimeSuite() || isScopedAgentsSuite() || isProjectSkillSuite() || + isParallelReadSuite() || isSteerRunnerKillSuite() ) { state.formalConfigPathReportLeakCount = countExactSecrets( @@ -1206,6 +1276,9 @@ try { const remainingProjectSkillReportLeakCount = isProjectSkillSuite() ? countExactSecrets(Buffer.from(report), state.projectSkill.privateValues) : 0; + const remainingParallelReadReportLeakCount = isParallelReadSuite() + ? countExactSecrets(Buffer.from(report), state.parallelRead.privateValues) + : 0; const remainingFormalConfigPathReportLeakCount = isWebSearchSuite() || isContextCompactionSuite() || @@ -1213,6 +1286,7 @@ try { isUserInputRuntimeSuite() || isScopedAgentsSuite() || isProjectSkillSuite() || + isParallelReadSuite() || isSteerRunnerKillSuite() ? countExactSecrets(Buffer.from(report), formalConfigPathVariants()) : 0; @@ -1224,6 +1298,7 @@ try { remainingUserInputReportLeakCount > 0 || remainingScopedAgentsReportLeakCount > 0 || remainingProjectSkillReportLeakCount > 0 || + remainingParallelReadReportLeakCount > 0 || remainingFormalConfigPathReportLeakCount > 0 ) { state.status = 'FAIL'; @@ -1240,9 +1315,11 @@ try { ? 'scoped-agents-report-redaction-required' : remainingProjectSkillReportLeakCount > 0 ? 'project-skill-report-redaction-required' - : remainingFormalConfigPathReportLeakCount > 0 - ? 'formal-config-path-report-redaction-required' - : 'web-search-report-redaction-required', + : remainingParallelReadReportLeakCount > 0 + ? 'parallel-read-report-redaction-required' + : remainingFormalConfigPathReportLeakCount > 0 + ? 'formal-config-path-report-redaction-required' + : 'web-search-report-redaction-required', ); const safeSummary = { status: state.status, @@ -1260,6 +1337,7 @@ try { userInputReportLeakCount: remainingUserInputReportLeakCount, scopedAgentsReportLeakCount: remainingScopedAgentsReportLeakCount, projectSkillReportLeakCount: remainingProjectSkillReportLeakCount, + parallelReadReportLeakCount: remainingParallelReadReportLeakCount, formalConfigPathReportLeakCount: remainingFormalConfigPathReportLeakCount, }, @@ -4835,6 +4913,577 @@ async function collectPartialProjectSkillEvidence() { }; } +async function seedParallelReadDisposableProject() { + await seedDisposableProject(); + const corpusRoot = path.join(state.projectRoot, parallelReadCorpusPath); + await fs.mkdir(corpusRoot, { recursive: true }); + const fillerBody = `${'parallel-read-filler '.repeat(820)}\n`; + const writes = []; + for (let index = 0; index < parallelReadCorpusFileCount; index += 1) { + writes.push( + fs.writeFile( + path.join(corpusRoot, `${String(index).padStart(4, '0')}.txt`), + fillerBody, + ), + ); + } + const evidenceBody = [ + `alpha evidence ${parallelReadAlphaQuery}`, + `beta evidence ${parallelReadBetaQuery}`, + '', + ].join('\n'); + const repositoryInstructions = `# Parallel read real E2E\n\n- This project is read-only for the current audit. Do not write files, run commands, call MCP, use Git, or delegate.\n- Verify ${parallelReadAlphaQuery} and ${parallelReadBetaQuery} with two separate project-wide text searches scoped to ${parallelReadCorpusPath}.\n- Submit both independent searches together in one planning turn before drawing a conclusion.\n- After both observations arrive, answer briefly without quoting repository instructions.\n- Never read or expose .env, ${configFileName}, or .agent/private-secret.txt.\n`; + await Promise.all([ + ...writes, + fs.writeFile(path.join(corpusRoot, 'zzzz-evidence.txt'), evidenceBody), + fs.writeFile( + path.join(state.projectRoot, 'AGENTS.md'), + repositoryInstructions, + ), + ]); + state.parallelRead.privateValues = [repositoryInstructions, evidenceBody]; +} + +function buildParallelReadTaskPrompt() { + return `对当前项目做一次严格只读的双证据核验：同时确认 ${parallelReadAlphaQuery} 与 ${parallelReadBetaQuery} 是否分别存在于 ${parallelReadCorpusPath}。必须实际执行两项彼此独立的项目全文搜索，并在同一个 planning 轮次一起提交；收到两项 observation 后再简短回答各自是否找到。不要修改项目，不要执行命令、MCP、Git 或委派。`; +} + +function assertParallelReadTaskPrompt(task) { + assert( + task.includes(parallelReadAlphaQuery) && + task.includes(parallelReadBetaQuery) && + task.includes(parallelReadCorpusPath) && + task.includes('同一个 planning 轮次') && + task.includes('严格只读'), + 'parallel-read-task-boundary-missing', + ); + for (const forbidden of [ + 'project.search', + 'file.read', + 'submit_agent_tool_plan', + state.projectRoot, + ]) { + assert(!task.includes(forbidden), 'parallel-read-task-runtime-recipe-leak'); + } +} + +async function runParallelReadE2e() { + await ensureOwnedRunnerStableKillSupport(); + await seedParallelReadDisposableProject(); + state.cliBinary = await prepareCliBinary(); + await prepareIsolatedSuiteAppData(); + + const task = buildParallelReadTaskPrompt(); + assertParallelReadTaskPrompt(task); + state.initialTask = { + chars: [...task].length, + sha256: hashValue(task), + }; + state.initialRunId = requestedRunId; + state.initialSessionId = goalSessionId; + state.isolatedRunner.launchAttempted = true; + await runCli( + [ + '--agent-enqueue', + '--init', + state.projectRoot, + mainAgentId, + state.initialRunId, + task, + ], + { timeoutMs: 120_000 }, + ); + await claimOwnedRunner(); + const runtime = await waitForResponseRuntimeIdentity(); + assert( + runtime.agentId === mainAgentId && + runtime.runId === state.initialRunId && + runtime.sessionId === state.initialSessionId, + 'parallel-read-runtime-identity-invalid', + ); + state.identityStable = true; + + await driveParallelReadRuntimeToCompletion(); + state.evidence = await validateParallelReadEvidence(); + assert(state.evidence.secretLeakCount === 0, 'loaded-key-leak-detected'); +} + +async function driveParallelReadRuntimeToCompletion() { + const deadline = Date.now() + runTimeoutMs; + let quietPolls = 0; + while (Date.now() < deadline) { + const pending = await findPendingActions(); + assert(pending.length === 0, 'parallel-read-unexpected-confirmation'); + const [runtime, tasks, conversations] = await Promise.all([ + readRuntime(mainAgentId).catch(() => null), + readTaskSnapshot(), + readOptionalJsonl( + agentConversationPath(mainAgentId, state.initialSessionId), + ), + ]); + const task = tasks.latest.find( + (candidate) => + candidate.agentId === mainAgentId && + candidate.runId === state.initialRunId, + ); + if (task && isFailedTask(task)) { + throw codedError('parallel-read-runtime-failed'); + } + if (runtime?.phase === 'needs-reconciliation') { + throw codedError('parallel-read-runtime-needs-reconciliation'); + } + const completed = + runtime?.runId === state.initialRunId && + runtime?.sessionId === state.initialSessionId && + runtime?.status === 'idle' && + runtime?.phase === 'completed' && + task?.status === 'completed' && + task?.phase === 'completed' && + conversations.filter((message) => message.role === 'assistant').length === + 1; + if (completed) { + quietPolls += 1; + if (quietPolls >= 3) return; + } else { + quietPolls = 0; + } + await sleep(250); + } + throw codedError('parallel-read-runtime-timeout'); +} + +function validateParallelReadProviderLifecycle(agentDb) { + const lifecycle = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.provider_request.lifecycle' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + const byRequest = new Map(); + for (const record of lifecycle) { + assert( + record.auditSchemaVersion === providerRequestLifecycleSchemaVersion && + isNonEmptyString(record.requestId) && + isNonEmptyString(record.requestSlot) && + ['tool-plan', 'final-reply'].includes(record.requestKind) && + record.webSearchEnabled === false, + 'parallel-read-provider-lifecycle-record-invalid', + ); + const records = byRequest.get(record.requestId) ?? []; + records.push(record); + byRequest.set(record.requestId, records); + } + for (const records of byRequest.values()) { + assert( + records.length === 2 && + records[0].status === 'started' && + records[1].status === 'completed' && + records[0].requestKind === records[1].requestKind && + records[0].requestSlot === records[1].requestSlot, + 'parallel-read-provider-lifecycle-sequence-invalid', + ); + } + const started = lifecycle.filter((record) => record.status === 'started'); + assert( + byRequest.size > 0 && + started.length === byRequest.size && + started.some((record) => record.requestKind === 'tool-plan'), + 'parallel-read-provider-request-count-invalid', + ); + return { + requestIdentityCount: byRequest.size, + startedCount: started.length, + terminalCount: lifecycle.filter((record) => record.status === 'completed') + .length, + toolPlanCount: started.filter( + (record) => record.requestKind === 'tool-plan', + ).length, + finalReplyCount: started.filter( + (record) => record.requestKind === 'final-reply', + ).length, + duplicateCount: duplicateCount( + lifecycle.map((record) => `${record.requestId}:${record.status}`), + ), + }; +} + +async function validateParallelReadEvidence() { + const persistence = await readScopedAgentsPersistence(); + const { + taskSnapshot, + events, + agentDb, + conversations, + activity, + output, + runtimeState, + contextBundle, + } = persistence; + const latest = taskSnapshot.latest.find( + (task) => task.agentId === mainAgentId && task.runId === state.initialRunId, + ); + assert( + runtimeState?.agentId === mainAgentId && + runtimeState.runId === state.initialRunId && + runtimeState.sessionId === state.initialSessionId && + runtimeState.status === 'idle' && + runtimeState.phase === 'completed' && + latest?.status === 'completed' && + latest.phase === 'completed', + 'parallel-read-final-runtime-invalid', + ); + + const batchAudits = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.parallel_read_batch.completed' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + assert(batchAudits.length === 1, 'parallel-read-batch-audit-count-invalid'); + const batch = batchAudits[0]; + assert( + isNonEmptyString(batch.batchId) && + batch.actionCount === 2 && + Array.isArray(batch.actionIds) && + batch.actionIds.length === 2 && + new Set(batch.actionIds).size === 2 && + Array.isArray(batch.tools) && + batch.tools.length === 2 && + batch.tools.every((tool) => tool === 'project.search') && + batch.overlapped === true && + Number.isFinite(batch.overlapNanos) && + batch.overlapNanos > 0, + 'parallel-read-batch-overlap-evidence-invalid', + ); + const batchActionIds = batch.actionIds; + const receipts = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.action_receipt' && + record.agentId === mainAgentId && + record.runId === state.initialRunId && + batchActionIds.includes(record.actionId), + ); + assert( + receipts.length === 2 && + receipts.every( + (record) => record.tool === 'project.search' && record.status === 'ok', + ) && + JSON.stringify(receipts.map((record) => record.actionId)) === + JSON.stringify(batchActionIds), + 'parallel-read-receipt-order-invalid', + ); + for (const actionId of batchActionIds) { + for (const recordType of [ + 'agent.runtime.tool_action.executing', + 'agent.runtime.tool_action.observed', + 'agent.runtime.action_receipt', + 'agent.runtime.tool_observation', + ]) { + const matching = agentDb.filter( + (record) => + record.recordType === recordType && + record.agentId === mainAgentId && + record.runId === state.initialRunId && + record.actionId === actionId, + ); + assert( + matching.length === 1, + `parallel-read-action-lifecycle-invalid:${recordType}`, + ); + if (recordType === 'agent.runtime.tool_observation') { + assert( + matching[0].parallelBatchId === batch.batchId && + matching[0].status === 'ok', + 'parallel-read-observation-batch-identity-invalid', + ); + } + } + } + const projectedCalls = (runtimeState.recentToolCalls ?? []).filter((call) => + batchActionIds.includes(call.actionId), + ); + assert( + projectedCalls.length === 2 && + JSON.stringify(projectedCalls.map((call) => call.actionId)) === + JSON.stringify(batchActionIds) && + projectedCalls.every( + (call) => call.tool === 'project.search' && call.status === 'ok', + ), + 'parallel-read-runtime-projection-order-invalid', + ); + const projectedDetails = projectedCalls.map((call) => call.detail ?? ''); + assert( + projectedDetails.filter((detail) => detail.includes(parallelReadAlphaQuery)) + .length === 1 && + projectedDetails.filter((detail) => + detail.includes(parallelReadBetaQuery), + ).length === 1, + 'parallel-read-independent-search-observation-missing', + ); + const receiptActionIds = receipts.map((record) => record.actionId); + const observationActionIds = agentDb + .filter( + (record) => + record.recordType === 'agent.runtime.tool_observation' && + record.agentId === mainAgentId && + record.runId === state.initialRunId && + batchActionIds.includes(record.actionId), + ) + .map((record) => record.actionId); + assert( + JSON.stringify(receiptActionIds) === JSON.stringify(batchActionIds) && + JSON.stringify(observationActionIds) === JSON.stringify(batchActionIds), + 'parallel-read-provider-order-projection-invalid', + ); + const batchEvents = events.filter( + (event) => + event.agentId === mainAgentId && + event.runId === state.initialRunId && + event.eventType === 'parallel_read_batch.completed', + ); + assert(batchEvents.length === 1, 'parallel-read-completed-event-invalid'); + + const protocolEvidence = + validateNativeRuntimeToolPlanProtocolEvidence(agentDb); + const nativeProtocols = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.tool_plan.protocol' && + record.agentId === mainAgentId && + record.runId === state.initialRunId && + record.protocol === 'native_runtime_tools', + ); + assert( + nativeProtocols.some( + (record) => + Number.isSafeInteger(record.functionCallCount) && + record.functionCallCount >= 2, + ), + 'parallel-read-native-multi-call-plan-missing', + ); + const providerLifecycle = validateParallelReadProviderLifecycle(agentDb); + assert( + providerLifecycle.duplicateCount === 0, + 'parallel-read-duplicate-provider-lifecycle', + ); + + const userMessages = conversations.filter( + (message) => message.role === 'user', + ); + const assistantMessages = conversations.filter( + (message) => message.role === 'assistant', + ); + const completedAudits = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.completed' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + const backgroundCompletedAudits = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.background_task.completed' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + assert( + userMessages.length === 1 && + assistantMessages.length === 1 && + completedAudits.length === 1 && + backgroundCompletedAudits.length === 1, + 'parallel-read-conversation-completion-cardinality-invalid', + ); + const duplicateMessageCount = duplicateCount( + conversations.map((message) => message.messageId).filter(Boolean), + ); + const duplicateReceiptCount = duplicateCount( + receipts.map(receiptAuditIdentity), + ); + const actionLifecycleRecords = agentDb.filter( + (record) => + batchActionIds.includes(record.actionId) && + [ + 'agent.runtime.tool_action.executing', + 'agent.runtime.tool_action.observed', + 'agent.runtime.action_receipt', + 'agent.runtime.tool_observation', + ].includes(record.recordType), + ); + const duplicateActionLifecycleCount = duplicateCount( + actionLifecycleRecords.map(actionAuditIdentity), + ); + assert( + duplicateMessageCount === 0 && + duplicateReceiptCount === 0 && + duplicateActionLifecycleCount === 0, + 'parallel-read-duplicate-persistence-identity', + ); + const finalizationFiles = ( + await listFiles( + path.join(state.projectRoot, '.agent/runtime/finalizations'), + ) + ).filter((file) => file.endsWith('.json')); + const pendingBatchFiles = ( + await listFiles( + path.join(state.projectRoot, '.agent/runtime/parallel-read-batches'), + ) + ).filter((file) => file.endsWith('.json')); + assert( + finalizationFiles.length === 0 && pendingBatchFiles.length === 0, + 'parallel-read-terminal-sidecar-present', + ); + + const publicSurfaces = { + task: taskSnapshot.all, + event: events, + agentDb, + activity, + output, + runtimeState, + contextBundle, + }; + const privateBodyPublicCounts = countSensitiveValuesBySurface( + { task: taskSnapshot.all, event: events, agentDb, activity, output }, + state.parallelRead.privateValues, + 'parallel-read-private-body-public', + ); + const apiKeyPublicCounts = countSensitiveValuesBySurface( + publicSurfaces, + state.secrets, + 'parallel-read-api-key-public', + ); + const projectPathPublicCounts = countSensitiveValuesBySurface( + publicSurfaces, + disposableProjectPathVariants(), + 'parallel-read-project-path-public', + ); + state.lureLeakCount = await countLureLeaks(); + assert(state.lureLeakCount === 0, 'sensitive-lure-leak-detected'); + const projectSecretLeakCount = await countSecretsInProject( + state.projectRoot, + state.secrets, + ); + const secretLeakCount = + (state.transcriptScanner?.count ?? 0) + projectSecretLeakCount; + assert(secretLeakCount === 0, 'loaded-key-leak-detected'); + + return { + scenario: 'native-tool-plan-persistent-parallel-read-batch', + targetAgentId: mainAgentId, + providerModel: state.parallelRead.effectiveModel, + providerApiKind: state.parallelRead.effectiveApiKind, + isolatedAppDataUsed: true, + formalConfigCliCallCount: state.isolatedRunner.sourceConfigCliCallCount, + sourceRunnerEndpointUnchanged: false, + sourceConfigReplicaCount: state.isolatedRunner.configLinks.length, + sourceConfigReplicasVerified: false, + taskCount: taskSnapshot.all.length, + eventCount: events.length, + agentDbRecordCount: agentDb.length, + conversationMessageCount: conversations.length, + targetRunCount: new Set( + taskSnapshot.all + .filter((task) => task.agentId === mainAgentId) + .map((task) => task.runId), + ).size, + stableSessionCount: new Set( + taskSnapshot.all + .filter((task) => task.agentId === mainAgentId) + .map((task) => task.sessionId), + ).size, + corpusFileCount: parallelReadCorpusFileCount + 1, + successfulSearchCount: receipts.length, + parallelBatchCount: batchAudits.length, + parallelBatchActionCount: batch.actionCount, + parallelBatchOverlapNanos: batch.overlapNanos, + parallelBatchOverlapped: batch.overlapped, + providerOrderedProjection: true, + ...protocolEvidence, + providerRequestIdentityCount: providerLifecycle.requestIdentityCount, + providerLifecycleStartedCount: providerLifecycle.startedCount, + providerLifecycleTerminalCount: providerLifecycle.terminalCount, + toolPlanProviderRequestCount: providerLifecycle.toolPlanCount, + finalReplyProviderRequestCount: providerLifecycle.finalReplyCount, + finalAssistantCount: assistantMessages.length, + completedAuditCount: completedAudits.length, + duplicateActionLifecycleCount, + duplicateReceiptCount, + duplicateProviderLifecycleCount: providerLifecycle.duplicateCount, + finalizationJournalCount: finalizationFiles.length, + privateBodyPublicLeakCount: sumObjectValues(privateBodyPublicCounts), + apiKeyPublicLeakCount: sumObjectValues(apiKeyPublicCounts), + projectPathPublicLeakCount: sumObjectValues(projectPathPublicCounts), + projectPathPublicSurfaceCount: Object.keys(projectPathPublicCounts).length, + parallelReadReportLeakCount: state.parallelRead.reportLeakCount, + parallelReadRunnerKillMethod: null, + parallelReadRunnerPidfdClaimCount: state.isolatedRunner.pidfdClaimCount, + parallelReadRunnerPidfdSignalCount: state.isolatedRunner.pidfdSignalCount, + parallelReadRunnerStopped: false, + parallelReadAppDataCleanupPerformed: false, + secretLeakCount, + lureLeakCount: state.lureLeakCount, + paths: [ + '.agent/runtime/context-bundles', + '.agent/runtime/tasks', + '.agent/runtime/events', + '.agent/agent.db', + '.agent/conversations', + ], + }; +} + +async function collectPartialParallelReadEvidence() { + const persistence = await readScopedAgentsPersistence(); + const { taskSnapshot, agentDb, conversations } = persistence; + const batches = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.parallel_read_batch.completed' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + const batchActionIds = batches.flatMap((record) => record.actionIds ?? []); + const receipts = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.action_receipt' && + record.agentId === mainAgentId && + record.runId === state.initialRunId && + batchActionIds.includes(record.actionId), + ); + const lifecycle = agentDb.filter( + (record) => + record.recordType === 'agent.runtime.provider_request.lifecycle' && + record.agentId === mainAgentId && + record.runId === state.initialRunId, + ); + const protocols = collectNativeRuntimeToolPlanProtocolEvidence(agentDb); + return { + providerModel: state.parallelRead.effectiveModel, + providerApiKind: state.parallelRead.effectiveApiKind, + taskCount: taskSnapshot.all.length, + eventCount: persistence.events.length, + agentDbRecordCount: agentDb.length, + conversationMessageCount: conversations.length, + successfulSearchCount: receipts.filter( + (record) => record.tool === 'project.search' && record.status === 'ok', + ).length, + parallelBatchCount: batches.length, + parallelBatchActionCount: batches[0]?.actionCount ?? 0, + parallelBatchOverlapNanos: batches[0]?.overlapNanos ?? 0, + parallelBatchOverlapped: batches[0]?.overlapped === true, + ...protocols, + providerRequestIdentityCount: new Set( + lifecycle.map((record) => record.requestId).filter(Boolean), + ).size, + providerLifecycleStartedCount: lifecycle.filter( + (record) => record.status === 'started', + ).length, + providerLifecycleTerminalCount: lifecycle.filter( + (record) => record.status === 'completed', + ).length, + finalAssistantCount: conversations.filter( + (message) => message.role === 'assistant', + ).length, + }; +} + async function runProcessSessionE2e() { await seedProcessSessionDisposableProject(); state.cliBinary = await prepareCliBinary(); @@ -4902,6 +5551,7 @@ function parseArguments(args) { suite === userInputRuntimeSuite || suite === scopedAgentsSuite || suite === projectSkillSuite || + suite === parallelReadSuite || suite === steerRunnerKillSuite || processSessionSuites.has(suite), 'unsupported-suite', @@ -5096,6 +5746,14 @@ function isolatedSuiteAppDataProfile() { codePrefix: 'project-skill-appdata', }; } + if (isParallelReadSuite()) { + return { + prefix: '.agent-runtime-real-e2e-parallel-read-', + sentinelName: parallelReadAppDataSentinelFileName, + sentinelSchema: parallelReadAppDataSentinelSchema, + codePrefix: 'parallel-read-appdata', + }; + } assert( isResponseStreamSuite(), 'isolated-appdata-used-outside-isolated-suite', @@ -5322,7 +5980,8 @@ async function prepareIsolatedSuiteAppData({ isWebSearchSuite() || isMcpRuntimeSuite() || isScopedAgentsSuite() || - isProjectSkillSuite() + isProjectSkillSuite() || + isParallelReadSuite() ? 'private-copy' : 'hardlink'; try { @@ -5541,6 +6200,25 @@ async function prepareIsolatedSuiteAppData({ state.projectSkill.effectiveModel = isolatedEffective.model; state.projectSkill.effectiveApiKind = isolatedEffective.apiKind; } + if (isParallelReadSuite()) { + const isolatedConfig = await loadConfig(appDataDir); + const isolatedEffective = effectiveAgentLlmConfig( + isolatedConfig.config, + mainAgentId, + ); + assert( + isolatedEffective.model === 'gpt-5.5' && + isolatedEffective.apiKind === 'openai_chat' && + ['apiKey', 'baseUrl', 'model'].every( + (key) => + typeof isolatedEffective[key] === 'string' && + isolatedEffective[key].trim().length > 0, + ), + 'parallel-read-effective-openai-chat-gpt-5-5-config-invalid', + ); + state.parallelRead.effectiveModel = isolatedEffective.model; + state.parallelRead.effectiveApiKind = isolatedEffective.apiKind; + } if (isSteerRunnerKillSuite()) { const isolatedConfig = await loadConfig(appDataDir); const isolatedEffective = effectiveAgentLlmConfig( @@ -15865,6 +16543,7 @@ function buildSummary() { isUserInputRuntimeSuite() || isScopedAgentsSuite() || isProjectSkillSuite() || + isParallelReadSuite() || isSteerRunnerKillSuite() ? { formalConfigPathTranscriptLeakCount: @@ -16530,6 +17209,64 @@ function emptyProjectSkillEvidence() { }; } +function emptyParallelReadEvidence() { + return { + scenario: 'native-tool-plan-persistent-parallel-read-batch', + targetAgentId: mainAgentId, + providerModel: null, + providerApiKind: null, + isolatedAppDataUsed: false, + formalConfigCliCallCount: 0, + sourceRunnerEndpointUnchanged: false, + sourceConfigReplicaCount: 0, + sourceConfigReplicasVerified: false, + taskCount: 0, + eventCount: 0, + agentDbRecordCount: 0, + conversationMessageCount: 0, + targetRunCount: 0, + stableSessionCount: 0, + corpusFileCount: parallelReadCorpusFileCount + 1, + successfulSearchCount: 0, + parallelBatchCount: 0, + parallelBatchActionCount: 0, + parallelBatchOverlapNanos: 0, + parallelBatchOverlapped: false, + providerOrderedProjection: false, + toolPlanProtocolCount: 0, + nativeRuntimeToolPlanCount: 0, + toolPlanRepairCount: 0, + nativeRuntimeToolPlanRepairCount: 0, + wrapperToolPlanFallbackCount: 0, + textJsonToolPlanFallbackCount: 0, + toolPlanAuditPayloadLeakCount: 0, + providerRequestIdentityCount: 0, + providerLifecycleStartedCount: 0, + providerLifecycleTerminalCount: 0, + toolPlanProviderRequestCount: 0, + finalReplyProviderRequestCount: 0, + finalAssistantCount: 0, + completedAuditCount: 0, + duplicateActionLifecycleCount: 0, + duplicateReceiptCount: 0, + duplicateProviderLifecycleCount: 0, + finalizationJournalCount: 0, + privateBodyPublicLeakCount: 0, + apiKeyPublicLeakCount: 0, + projectPathPublicLeakCount: 0, + projectPathPublicSurfaceCount: 0, + parallelReadReportLeakCount: 0, + parallelReadRunnerKillMethod: null, + parallelReadRunnerPidfdClaimCount: 0, + parallelReadRunnerPidfdSignalCount: 0, + parallelReadRunnerStopped: false, + parallelReadAppDataCleanupPerformed: false, + secretLeakCount: 0, + lureLeakCount: 0, + paths: [], + }; +} + function emptyGoalEvidence() { return { scenario: 'goal-edit-pause-runner-restart-resume', @@ -17442,6 +18179,10 @@ function isProjectSkillSuite() { return state.suite === projectSkillSuite; } +function isParallelReadSuite() { + return state.suite === parallelReadSuite; +} + function isSteerRunnerKillSuite() { return state.suite === steerRunnerKillSuite; } @@ -17456,7 +18197,8 @@ function isIsolatedRunnerSuite() { isMcpRuntimeSuite() || isUserInputRuntimeSuite() || isScopedAgentsSuite() || - isProjectSkillSuite() + isProjectSkillSuite() || + isParallelReadSuite() ); } diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent.rs b/apps/ai-game-creator-shell/src-tauri/src/agent.rs index b58c18de8..35af51d20 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent.rs @@ -41,6 +41,11 @@ pub(crate) const AGENT_RUNTIME_VERIFICATION_GATE_SCHEMA_VERSION: &str = const AGENT_RUNTIME_PROJECT_REVISION_RELATIVE_PATH: &str = ".agent/runtime/project-revision.json"; const AGENT_RUNTIME_SIDECAR_MAX_BYTES: usize = 16 * 1024; const AGENT_RUNTIME_PENDING_ACTION_SIDECAR_MAX_BYTES: usize = 512 * 1024; +const AGENT_RUNTIME_PARALLEL_READ_BATCH_SCHEMA_VERSION: &str = + "game-creator-parallel-read-batch.v1"; +const AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING: &str = "executing"; +const AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED: &str = "observed"; +const AGENT_RUNTIME_PARALLEL_READ_BATCH_SIDECAR_MAX_BYTES: usize = 4 * 1024 * 1024; const AGENT_RUNTIME_FINALIZATION_SIDECAR_MAX_BYTES: usize = 512 * 1024; const AGENT_RUNTIME_RESPONSE_STREAM_SCHEMA_VERSION: &str = "game-creator-runtime-response-stream.v1"; @@ -679,6 +684,17 @@ pub(crate) fn resume_game_creator_agent_background_tasks_at( drop(runtime_lock); continue; } + let runtime_lock = match resume_game_creator_agent_parallel_read_batch_at( + root, + &agent_id, + runtime_lock, + )? { + AgentRuntimePendingActionResume::Handled(result) => { + resumed.push(result); + continue; + } + AgentRuntimePendingActionResume::NotFound(runtime_lock) => runtime_lock, + }; let runtime_lock = match resume_game_creator_agent_pending_tool_action_at( root, &agent_id, @@ -1698,6 +1714,130 @@ fn suppress_static_delegate_reservation_for_rejected_pending_action_at( suppress_static_delegate_delivery_at(root, &expected).map(|_| ()) } +fn resume_game_creator_agent_parallel_read_batch_at( + root: &Path, + agent_id: &str, + runtime_lock: AgentRuntimeTaskLock, +) -> Result { + let mut runtime = read_game_creator_agent_runtime_at(root, agent_id)?.state; + if runtime.run_id.trim().is_empty() + || !game_creator_agent_runtime_parallel_read_batch_exists(root, agent_id, &runtime.run_id) + { + return Ok(AgentRuntimePendingActionResume::NotFound(runtime_lock)); + } + let batch = match read_game_creator_agent_runtime_parallel_read_batch( + root, + agent_id, + &runtime.run_id, + ) { + Ok(batch) => batch, + Err(error) => { + let error = format!("Agent Runtime 只读并行批次恢复失败并已关闭当前 run：{error}"); + let failed = fail_game_creator_agent_runtime_turn_at(root, runtime, &error)?; + let _ = append_agent_db_record( + root, + serde_json::json!({ + "recordType": "agent.runtime.parallel_read_batch.recovery_failed", + "agentId": failed.agent_id, + "taskId": failed.task_id, + "sessionId": failed.session_id, + "runId": failed.run_id, + "source": failed.source, + "error": failed.error, + }), + ); + return read_game_creator_agent_runtime_at(root, agent_id) + .map(AgentRuntimePendingActionResume::Handled); + } + }; + let first_pending = batch + .actions + .first() + .ok_or_else(|| "Agent Runtime 只读并行批次缺少恢复动作".to_string())?; + if validate_agent_runtime_pending_context(root, &runtime, first_pending).is_err() + || batch.agent_id != runtime.agent_id + || batch.task_id != runtime.task_id + || batch.session_id != runtime.session_id + || batch.run_id != runtime.run_id + || batch.source != runtime.source + { + mark_game_creator_agent_runtime_needs_reconciliation_at( + root, + &mut runtime, + first_pending, + "Agent Runtime 只读并行批次与当前状态身份不一致", + )?; + return read_game_creator_agent_runtime_at(root, agent_id) + .map(AgentRuntimePendingActionResume::Handled); + } + if matches!(runtime.phase.as_str(), "completed" | "cancelled" | "failed") + && runtime.phase != "needs-reconciliation" + { + remove_game_creator_agent_runtime_parallel_read_batch(root, agent_id, &runtime.run_id)?; + return Ok(AgentRuntimePendingActionResume::NotFound(runtime_lock)); + } + if runtime.phase == "needs-reconciliation" + || game_creator_agent_runtime_has_reconciliation_barrier(root, agent_id)? + { + return read_game_creator_agent_runtime_at(root, agent_id) + .map(AgentRuntimePendingActionResume::Handled); + } + if batch.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING + && game_creator_agent_runtime_cancel_requested(root, &runtime) + { + remove_game_creator_agent_runtime_parallel_read_batch(root, agent_id, &runtime.run_id)?; + mark_game_creator_agent_runtime_cancelled_at( + root, + &mut runtime, + "Agent 后台任务已按开发者请求取消", + Some("Runtime 恢复只读并行批次时发现尚未完成的取消请求。"), + )?; + return read_game_creator_agent_runtime_at(root, agent_id) + .map(AgentRuntimePendingActionResume::Handled); + } + runtime.status = "running".to_string(); + runtime.phase = if batch.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING { + "action".to_string() + } else { + "observation".to_string() + }; + runtime.current_action = format!("恢复 {} 个只读工具的持久并行批次", batch.actions.len()); + runtime.waiting_on = "Runtime 恢复只读并行批次".to_string(); + runtime.next_step = if batch.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING { + "在同一 action identity 下安全重放未落盘的只读观察".to_string() + } else { + "按 Provider 顺序补齐只读观察投影".to_string() + }; + runtime.pending_tool_action = None; + runtime.error = None; + runtime.updated_at = unix_timestamp(); + append_game_creator_agent_runtime_task(root, &runtime)?; + refresh_game_creator_agent_runtime_task_queue(root, &mut runtime)?; + write_game_creator_agent_runtime_state(root, &runtime)?; + append_game_creator_agent_runtime_event( + root, + &runtime, + "parallel_read_batch.resume", + "running", + runtime.phase.as_str(), + "Runtime 正在同一 run 恢复持久只读并行批次。", + Some(&format!( + "batchId={} · actionCount={} · status={}", + batch.batch_id, + batch.actions.len(), + batch.status + )), + )?; + let result = read_game_creator_agent_runtime_at(root, agent_id)?; + let root = root.to_path_buf(); + let agent_id = agent_id.to_string(); + tauri::async_runtime::spawn(async move { + let _runtime_lock = runtime_lock; + continue_game_creator_agent_parallel_read_batch(root, agent_id, batch, runtime).await; + }); + Ok(AgentRuntimePendingActionResume::Handled(result)) +} + fn resume_game_creator_agent_pending_tool_action_at( root: &Path, agent_id: &str, @@ -3100,7 +3240,7 @@ fn resolve_game_creator_agent_runtime_cancel_target( }) .ok_or_else(|| format!("未找到 Agent Runtime 任务：{run_id}"))?; let has_pending_action = - game_creator_agent_runtime_pending_tool_action_exists(root, agent_id, run_id); + game_creator_agent_runtime_has_pending_action_ledger(root, agent_id, run_id); Ok((current_result, task, has_pending_action)) } @@ -3187,7 +3327,7 @@ pub(crate) fn retry_game_creator_agent_runtime_task_at( return Err("持久 Goal 任务不能使用普通 retry；请清理后创建新 Goal".to_string()); } if task.phase == "needs-reconciliation" - || game_creator_agent_runtime_pending_tool_action_exists(root, &agent_id, &target_run_id) + || game_creator_agent_runtime_has_pending_action_ledger(root, &agent_id, &target_run_id) { return Err("Agent Runtime 仍保留待核对工具动作，请先核对项目状态并取消原任务".to_string()); } @@ -3797,6 +3937,239 @@ pub(crate) fn answer_game_creator_agent_runtime_user_input_at( Ok(result) } +fn blocked_game_creator_agent_runtime_parallel_read_results( + batch: &AgentRuntimeParallelReadBatch, + summary: &str, +) -> Vec<(AgentRuntimeToolObservation, AgentRuntimeParallelReadTiming)> { + let timestamp = agent_runtime_timestamp_nanos_u64(); + batch + .actions + .iter() + .map(|pending| { + ( + AgentRuntimeToolObservation { + tool: pending.action.tool.clone(), + status: "blocked".to_string(), + summary: summary.to_string(), + detail: Some(format!( + "parallelBatchId={} · plannedSteerCursor={}", + batch.batch_id, batch.planned_steer_cursor + )), + }, + AgentRuntimeParallelReadTiming { + action_id: pending.action_id.clone(), + started_at_nanos: timestamp, + finished_at_nanos: timestamp, + }, + ) + }) + .collect() +} + +fn recover_game_creator_agent_runtime_parallel_read_batch_blocking( + root: PathBuf, + expected: AgentRuntimeParallelReadBatch, +) -> Result { + if expected.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED { + return Ok(expected); + } + let _project_lock = acquire_game_creator_agent_runtime_project_write_lock_with_wait( + &root, + "runtime.parallel_read_batch.recover", + )?; + let durable = read_game_creator_agent_runtime_parallel_read_batch( + &root, + &expected.agent_id, + &expected.run_id, + )?; + if durable != expected { + return Err("Agent Runtime 只读并行批次恢复时 durable 内容已变化".to_string()); + } + let runtime = read_game_creator_agent_runtime_for_session_at( + &root, + &expected.agent_id, + Some(&expected.session_id), + )? + .state; + let control_stale = game_creator_agent_runtime_cancel_requested(&root, &runtime) + || game_creator_agent_runtime_has_queued_steer_after_cursor( + &root, + &expected.agent_id, + &expected.run_id, + expected.planned_steer_cursor, + )?; + let mut gate_stale = false; + for pending in &expected.actions { + if validate_game_creator_agent_runtime_parallel_read_pending_at_locked( + &root, &runtime, pending, + )? + .is_some() + { + gate_stale = true; + break; + } + } + if control_stale || gate_stale { + return mark_game_creator_agent_runtime_parallel_read_batch_observed_at_locked( + &root, + expected.clone(), + blocked_game_creator_agent_runtime_parallel_read_results( + &expected, + "Runner 恢复时控制或一致性快照已变化，旧只读批次未重放", + ), + ); + } + execute_game_creator_agent_runtime_parallel_read_members_at_locked(&root, expected) +} + +async fn recover_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + batch: AgentRuntimeParallelReadBatch, +) -> Result { + let root = root.to_path_buf(); + tokio::task::spawn_blocking(move || { + recover_game_creator_agent_runtime_parallel_read_batch_blocking(root, batch) + }) + .await + .map_err(|error| format!("Agent Runtime 只读并行批次恢复调度失败：{error}"))? +} + +async fn continue_game_creator_agent_parallel_read_batch( + root: PathBuf, + agent_id: String, + batch: AgentRuntimeParallelReadBatch, + mut runtime: AgentRuntimeState, +) { + let first_pending = match batch.actions.first() { + Some(pending) => pending.clone(), + None => return, + }; + if let Err(error) = validate_agent_runtime_pending_context(&root, &runtime, &first_pending) { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &error, + ); + return; + } + let batch = match recover_game_creator_agent_runtime_parallel_read_batch(&root, batch).await { + Ok(batch) => batch, + Err(error) => { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!("恢复只读并行批次失败：{error}"), + ); + return; + } + }; + let context_bundle = match read_game_creator_agent_runtime_context_bundle_with_superseded_goal( + &root, + &runtime, + Some(&first_pending), + false, + ) { + Ok(bundle) => bundle, + Err(error) => { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!("恢复只读并行批次 context bundle 失败：{error}"), + ); + return; + } + }; + let mut continuation = context_bundle + .map(continuation_from_game_creator_agent_runtime_context_bundle) + .unwrap_or_default(); + let mut observations = first_pending.observations.clone(); + let mut plan = first_pending.tool_plan(); + let next_loop_index = usize::try_from(first_pending.loop_iteration).unwrap_or(usize::MAX); + let mut context_tracker = AgentRuntimeContextWindowTracker::from_continuation(&continuation); + if let Err(error) = project_game_creator_agent_runtime_parallel_read_batch( + &root, + &mut runtime, + &first_pending.task, + &plan, + &mut observations, + next_loop_index, + &mut context_tracker, + &batch, + ) { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!("恢复只读并行批次 observation 投影失败：{error}"), + ); + return; + } + if stop_game_creator_agent_runtime_if_cancel_requested(&root, &mut runtime) { + drain_next_game_creator_agent_background_tasks(root, agent_id).await; + return; + } + match consume_game_creator_agent_runtime_steers( + &root, + &mut runtime, + &first_pending.task, + &plan, + &mut observations, + next_loop_index, + &context_tracker, + ) { + Ok(true) => plan = AgentRuntimeToolPlan::default(), + Ok(false) => {} + Err(error) => { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!("恢复只读并行批次后消费 steer 失败：{error}"), + ); + return; + } + } + let checkpoint = match checkpoint_game_creator_agent_runtime_context( + &root, + &mut runtime, + &first_pending.task, + &plan, + &mut observations, + next_loop_index, + &mut context_tracker, + ) { + Ok(checkpoint) => checkpoint, + Err(error) => { + let _ = mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!("恢复只读并行批次 checkpoint 失败：{error}"), + ); + return; + } + }; + continuation.plan = plan; + continuation.observations = observations; + continuation.next_loop_index = next_loop_index; + continuation.context_stalled = checkpoint == AgentRuntimeContextCheckpoint::Stalled; + context_tracker.apply_to_continuation(&mut continuation); + let outcome = run_game_creator_agent_background_task_with_context( + root.clone(), + agent_id.clone(), + first_pending.task.clone(), + runtime, + continuation, + ) + .await; + if matches!(outcome, AgentBackgroundTaskOutcome::Finished) { + drain_next_game_creator_agent_background_tasks(root, agent_id).await; + } +} + pub(crate) async fn continue_game_creator_agent_pending_tool_action( root: PathBuf, agent_id: String, @@ -5887,12 +6260,216 @@ async fn run_game_creator_agent_background_task_pass_with_context( } let mut steered_during_actions = false; + let mut parallel_batch_consumed_until = 0_usize; for (action_index, action) in plan .actions .iter() .take(AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT) .enumerate() { + if action_index < parallel_batch_consumed_until { + continue; + } + let parallel_batch_len = agent_runtime_parallel_read_batch_len( + &plan.actions[..plan + .actions + .len() + .min(AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT)], + action_index, + ); + if parallel_batch_len >= 2 { + let batch_actions = plan.actions + [action_index..action_index.saturating_add(parallel_batch_len)] + .to_vec(); + if agent_runtime_parallel_read_batch_is_auto_at(&root, &agent_id, &batch_actions) { + activate_agent_runtime_plan_step( + &mut runtime, + action_index, + action.reason.as_deref().unwrap_or(action.tool.as_str()), + ); + let fallback = runtime.clone(); + runtime = match advance_game_creator_agent_runtime_turn_at( + &root, + runtime, + "action", + &format!("并行读取 {} 个独立项目上下文", parallel_batch_len), + "Agent 请求了一组可并行的只读工具动作。", + ) { + Ok(runtime) => runtime, + Err(error) => { + let _ = + fail_game_creator_agent_runtime_turn_at(&root, fallback, &error); + return AgentBackgroundTaskOutcome::Finished; + } + }; + let _ = append_game_creator_agent_runtime_event( + &root, + &runtime, + "action", + runtime.status.as_str(), + runtime.phase.as_str(), + runtime.current_action.as_str(), + Some(&format!( + "parallelReadBatch=true · actionCount={parallel_batch_len}" + )), + ); + if let Err(error) = persist_game_creator_agent_runtime_context( + &root, + &runtime, + &task, + &plan, + &observations, + loop_index, + &context_tracker, + ) { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &error, + ); + } + if stop_game_creator_agent_runtime_if_cancel_requested(&root, &mut runtime) { + return AgentBackgroundTaskOutcome::Finished; + } + if let Err(error) = + clear_game_creator_agent_runtime_observed_action_ledger(&root, &mut runtime) + { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("建立只读并行批次前清理已完成工具动作账本失败：{error}"), + ); + } + match execute_game_creator_agent_runtime_parallel_read_batch( + &root, + &runtime, + &task, + &plan, + &observations, + &planning_request_revision, + &planning_repository_context_fingerprint, + action_index, + &batch_actions, + ) + .await + { + Ok(AgentRuntimeParallelReadBatchExecution::Executed(batch)) => { + let first_pending = batch + .actions + .first() + .expect("parallel read batch has at least two actions") + .clone(); + let repository_context_drifted = + match project_game_creator_agent_runtime_parallel_read_batch( + &root, + &mut runtime, + &task, + &plan, + &mut observations, + loop_index, + &mut context_tracker, + &batch, + ) { + Ok(drifted) => drifted, + Err(error) => { + let _ = + mark_game_creator_agent_runtime_needs_reconciliation_at( + &root, + &mut runtime, + &first_pending, + &format!( + "只读并行批次 observation 投影失败：{error}" + ), + ); + return AgentBackgroundTaskOutcome::NeedsReconciliation; + } + }; + parallel_batch_consumed_until = + action_index.saturating_add(parallel_batch_len); + if stop_game_creator_agent_runtime_if_cancel_requested( + &root, + &mut runtime, + ) { + return AgentBackgroundTaskOutcome::Finished; + } + match consume_game_creator_agent_runtime_steers( + &root, + &mut runtime, + &task, + &plan, + &mut observations, + loop_index + 1, + &context_tracker, + ) { + Ok(true) => { + steered_during_actions = true; + break; + } + Ok(false) => {} + Err(error) => { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("只读并行批次后消费追加指令失败：{error}"), + ); + } + } + if repository_context_drifted { + break; + } + continue; + } + Ok(AgentRuntimeParallelReadBatchExecution::Stale) => { + if stop_game_creator_agent_runtime_if_cancel_requested( + &root, + &mut runtime, + ) { + return AgentBackgroundTaskOutcome::Finished; + } + match consume_game_creator_agent_runtime_steers( + &root, + &mut runtime, + &task, + &plan, + &mut observations, + loop_index + 1, + &context_tracker, + ) { + Ok(true) => { + steered_during_actions = true; + break; + } + Ok(false) => {} + Err(error) => { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("只读并行批次建立前消费追加指令失败：{error}"), + ); + } + } + } + Ok(AgentRuntimeParallelReadBatchExecution::NotEligible) => {} + Err(error) => { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("执行只读并行批次失败：{error}"), + ); + } + } + } + } activate_agent_runtime_plan_step( &mut runtime, action_index, @@ -7003,6 +7580,13 @@ pub(crate) enum AgentBackgroundTaskOutcome { }, } +#[derive(Debug)] +enum AgentRuntimeParallelReadBatchExecution { + Executed(AgentRuntimeParallelReadBatch), + NotEligible, + Stale, +} + #[derive(Debug)] pub(crate) enum AgentBackgroundFinalizationOutcome { Completed(AgentRuntimeState), @@ -8601,7 +9185,7 @@ fn game_creator_agent_runtime_provider_snapshot_has_durable_control_at_locked( return Ok(runtime.status != "idle" || !matches!(runtime.phase.as_str(), "idle" | "completed") || runtime.pending_tool_action.is_some() - || game_creator_agent_runtime_pending_tool_action_exists( + || game_creator_agent_runtime_has_pending_action_ledger( root, &runtime.agent_id, &runtime.run_id, @@ -8659,7 +9243,7 @@ fn game_creator_agent_runtime_provider_snapshot_has_durable_control_at_locked( "idle" | "completed" | "failed" | "cancelled" ) && runtime.pending_tool_action.is_none() - && !game_creator_agent_runtime_pending_tool_action_exists( + && !game_creator_agent_runtime_has_pending_action_ledger( root, &runtime.agent_id, &runtime.run_id, @@ -11338,6 +11922,34 @@ pub(crate) struct AgentRuntimePendingToolAction { pub(crate) updated_at: u64, } +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct AgentRuntimeParallelReadTiming { + action_id: String, + started_at_nanos: u64, + finished_at_nanos: u64, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct AgentRuntimeParallelReadBatch { + schema_version: String, + batch_id: String, + project_id: String, + agent_id: String, + task_id: String, + session_id: String, + run_id: String, + source: String, + loop_iteration: u32, + planned_steer_cursor: u64, + status: String, + actions: Vec, + timings: Vec, + created_at: u64, + updated_at: u64, +} + impl AgentRuntimePendingToolAction { pub(crate) fn summary(&self) -> AgentRuntimePendingToolActionSummary { AgentRuntimePendingToolActionSummary { @@ -11606,6 +12218,937 @@ fn append_game_creator_agent_runtime_auto_tool_action_observed_record( ) } +fn append_game_creator_agent_runtime_parallel_action_executing_record( + root: &Path, + pending: &AgentRuntimePendingToolAction, +) -> Result<(), String> { + let record_type = "agent.runtime.tool_action.executing"; + if agent_db_record_exists_for_action( + root, + record_type, + &pending.agent_id, + &pending.run_id, + &pending.action_id, + )? { + return Ok(()); + } + append_game_creator_agent_runtime_auto_tool_action_executing_record(root, pending) +} + +fn append_game_creator_agent_runtime_parallel_action_observed_record( + root: &Path, + pending: &AgentRuntimePendingToolAction, + observation: &AgentRuntimeToolObservation, +) -> Result<(), String> { + let record_type = "agent.runtime.tool_action.observed"; + if agent_db_record_exists_for_action( + root, + record_type, + &pending.agent_id, + &pending.run_id, + &pending.action_id, + )? { + return Ok(()); + } + append_game_creator_agent_runtime_auto_tool_action_observed_record(root, pending, observation) +} + +fn agent_runtime_timestamp_nanos_u64() -> u64 { + unix_timestamp_nanos().min(u128::from(u64::MAX)) as u64 +} + +fn agent_runtime_parallel_read_overlap_nanos(timings: &[AgentRuntimeParallelReadTiming]) -> u64 { + let mut maximum_overlap = 0_u64; + for (index, left) in timings.iter().enumerate() { + for right in timings.iter().skip(index + 1) { + let overlap_start = left.started_at_nanos.max(right.started_at_nanos); + let overlap_end = left.finished_at_nanos.min(right.finished_at_nanos); + maximum_overlap = maximum_overlap.max(overlap_end.saturating_sub(overlap_start)); + } + } + maximum_overlap +} + +#[cfg(test)] +fn wait_on_agent_runtime_parallel_read_test_barrier(root: &Path) { + static BARRIERS: OnceLock< + std::sync::Mutex>>, + > = OnceLock::new(); + let marker = root.join(".agent/runtime/test-parallel-read-barrier-count"); + let count = fs::read_to_string(marker) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .filter(|count| (2..=AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT).contains(count)); + let Some(count) = count else { + return; + }; + let key = format!("{}\n{count}", root.to_string_lossy()); + let barrier = { + let mut barriers = BARRIERS + .get_or_init(|| std::sync::Mutex::new(std::collections::BTreeMap::new())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + barriers + .entry(key) + .or_insert_with(|| Arc::new(std::sync::Barrier::new(count))) + .clone() + }; + barrier.wait(); + let hold = root.join(".agent/runtime/test-parallel-read-hold"); + if hold.exists() { + let release = root.join(".agent/runtime/test-parallel-read-release"); + let started = std::time::Instant::now(); + while !release.exists() { + assert!( + started.elapsed() < Duration::from_secs(5), + "parallel read test release marker timed out" + ); + std::thread::sleep(Duration::from_millis(5)); + } + } +} + +#[cfg(not(test))] +fn wait_on_agent_runtime_parallel_read_test_barrier(_root: &Path) {} + +#[cfg(test)] +fn delay_agent_runtime_parallel_read_test_action( + root: &Path, + pending: &AgentRuntimePendingToolAction, +) { + if pending.action_index != 0 { + return; + } + let delay_millis = + fs::read_to_string(root.join(".agent/runtime/test-parallel-read-delay-first-ms")) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .filter(|value| *value <= 1_000) + .unwrap_or_default(); + if delay_millis > 0 { + std::thread::sleep(Duration::from_millis(delay_millis)); + } +} + +#[cfg(not(test))] +fn delay_agent_runtime_parallel_read_test_action( + _root: &Path, + _pending: &AgentRuntimePendingToolAction, +) { +} + +fn execute_game_creator_agent_runtime_parallel_safe_read_at_locked( + root: &Path, + agent_id: &str, + run_id: &str, + pending: &AgentRuntimePendingToolAction, +) -> AgentRuntimeToolObservation { + let action = &pending.action; + let tool = action.tool.trim(); + if agent_id.trim().starts_with("child-") { + if let Err(error) = + validate_isolated_agent_tool_scope_at(root, agent_id.trim(), tool, &action.input) + { + return AgentRuntimeToolObservation { + tool: tool.to_string(), + status: "rejected".to_string(), + summary: redact_agent_runtime_project_paths(root, &error, 240), + detail: None, + }; + } + } + match tool { + "memory.read" => observe_agent_runtime_memory(root, agent_id, &action.input), + "conversation.read" => observe_agent_runtime_conversation(root, agent_id, run_id), + "asset.list" => observe_agent_runtime_assets(root), + "project.search" => observe_agent_runtime_project_search(root, &action.input), + "project.diff" => observe_agent_runtime_project_diff(root, &action.input), + "git.inspect" => observe_agent_runtime_git_inspect(root, &action.input), + "file.list" => observe_agent_runtime_file_list(root, &action.input), + "file.read" => observe_agent_runtime_file(root, &action.input), + "task.list" => observe_agent_runtime_task_list(root), + _ => AgentRuntimeToolObservation { + tool: tool.to_string(), + status: "rejected".to_string(), + summary: "工具不属于 Agent Runtime 只读并行白名单".to_string(), + detail: None, + }, + } +} + +fn validate_game_creator_agent_runtime_parallel_read_pending_at_locked( + root: &Path, + runtime: &AgentRuntimeState, + pending: &AgentRuntimePendingToolAction, +) -> Result, String> { + validate_agent_runtime_pending_tool_action_record(root, pending)?; + validate_agent_runtime_pending_context(root, runtime, pending)?; + if !pending.is_auto() || !agent_runtime_tool_is_parallel_safe_read(&pending.action.tool) { + return Err("Agent Runtime 只读并行成员不属于自动只读动作".to_string()); + } + if let Err(error) = validate_agent_runtime_pending_current_goal_snapshot(root, pending) { + return Ok(Some(agent_runtime_pending_goal_stale_observation( + root, &error, + ))); + } + if let Some(observation) = pending_repository_context_drift_observation(root, pending)? { + return Ok(Some(observation)); + } + let Some(command_id) = game_creator_agent_runtime_tool_command_id(&pending.action.tool) else { + return Err("Agent Runtime 只读并行成员不在工具白名单中".to_string()); + }; + if let Some(blocked) = game_creator_agent_runtime_tool_policy_block_after_lock( + root, + &pending.agent_id, + command_id, + Some(pending), + ) { + return Ok(Some(agent_runtime_tool_policy_block_observation( + &pending.action.tool, + blocked, + ))); + } + Ok(None) +} + +fn mark_game_creator_agent_runtime_parallel_read_batch_observed_at_locked( + root: &Path, + mut batch: AgentRuntimeParallelReadBatch, + results: Vec<(AgentRuntimeToolObservation, AgentRuntimeParallelReadTiming)>, +) -> Result { + if results.len() != batch.actions.len() { + return Err("Agent Runtime 只读并行批次返回数量不匹配".to_string()); + } + batch.timings.clear(); + for (pending, (observation, timing)) in batch.actions.iter_mut().zip(results) { + if observation.tool != pending.action.tool + || observation.is_waiting_for_confirmation() + || observation.requires_reconciliation() + || timing.action_id != pending.action_id + { + return Err(format!( + "Agent Runtime 只读并行成员返回了不可持久化的 observation：{} / {}", + pending.action.tool, observation.status + )); + } + pending.status = AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED.to_string(); + pending.observation = Some(observation); + pending.updated_at = unix_timestamp(); + batch.timings.push(timing); + } + batch.status = AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED.to_string(); + batch.updated_at = unix_timestamp(); + write_game_creator_agent_runtime_parallel_read_batch(root, &batch)?; + for pending in &batch.actions { + if let Some(observation) = pending.observation.as_ref() { + let _ = append_game_creator_agent_runtime_parallel_action_observed_record( + root, + pending, + observation, + ); + } + } + Ok(batch) +} + +fn append_game_creator_agent_runtime_parallel_read_batch_completed_audit( + root: &Path, + batch: &AgentRuntimeParallelReadBatch, +) -> Result<(), String> { + let overlap_nanos = agent_runtime_parallel_read_overlap_nanos(&batch.timings); + let started_at_nanos = batch + .timings + .iter() + .map(|timing| timing.started_at_nanos) + .min() + .unwrap_or_default(); + let finished_at_nanos = batch + .timings + .iter() + .map(|timing| timing.finished_at_nanos) + .max() + .unwrap_or_default(); + let first_action_id = batch + .actions + .first() + .map(|pending| pending.action_id.as_str()) + .unwrap_or_default(); + let audit = serde_json::json!({ + "recordType": "agent.runtime.parallel_read_batch.completed", + "agentId": batch.agent_id, + "taskId": batch.task_id, + "sessionId": batch.session_id, + "runId": batch.run_id, + "batchId": batch.batch_id, + "actionId": first_action_id, + "actionIds": batch.actions.iter().map(|pending| pending.action_id.as_str()).collect::>(), + "tools": batch.actions.iter().map(|pending| pending.action.tool.as_str()).collect::>(), + "actionCount": batch.actions.len(), + "startedAtNanos": started_at_nanos, + "finishedAtNanos": finished_at_nanos, + "overlapNanos": overlap_nanos, + "overlapped": overlap_nanos > 0, + }); + if !agent_db_record_exists_for_action( + root, + "agent.runtime.parallel_read_batch.completed", + &batch.agent_id, + &batch.run_id, + first_action_id, + )? { + append_agent_db_record(root, audit)?; + } + Ok(()) +} + +fn execute_game_creator_agent_runtime_parallel_read_members_at_locked( + root: &Path, + batch: AgentRuntimeParallelReadBatch, +) -> Result { + let results = std::thread::scope(|scope| { + let handles = batch + .actions + .iter() + .map(|pending| { + scope.spawn(move || { + let started_at_nanos = agent_runtime_timestamp_nanos_u64(); + wait_on_agent_runtime_parallel_read_test_barrier(root); + delay_agent_runtime_parallel_read_test_action(root, pending); + let observation = + execute_game_creator_agent_runtime_parallel_safe_read_at_locked( + root, + &pending.agent_id, + &pending.run_id, + pending, + ); + let finished_at_nanos = agent_runtime_timestamp_nanos_u64(); + ( + observation, + AgentRuntimeParallelReadTiming { + action_id: pending.action_id.clone(), + started_at_nanos, + finished_at_nanos, + }, + ) + }) + }) + .collect::>(); + handles + .into_iter() + .map(|handle| { + handle + .join() + .map_err(|_| "Agent Runtime 只读并行工作线程异常退出".to_string()) + }) + .collect::, String>>() + })?; + mark_game_creator_agent_runtime_parallel_read_batch_observed_at_locked(root, batch, results) +} + +fn prepare_and_execute_game_creator_agent_runtime_parallel_read_batch_blocking( + root: PathBuf, + runtime: AgentRuntimeState, + task: String, + plan: AgentRuntimeToolPlan, + observations: Vec, + project_revision_before: AgentRuntimeProjectRevision, + repository_context_fingerprint: String, + action_start_index: usize, + actions: Vec, +) -> Result { + if actions.len() < 2 + || actions.len() > AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT + || !actions + .iter() + .all(|action| agent_runtime_tool_is_parallel_safe_read(&action.tool)) + { + return Ok(AgentRuntimeParallelReadBatchExecution::NotEligible); + } + let _project_lock = acquire_game_creator_agent_runtime_project_write_lock_with_wait( + &root, + "runtime.parallel_read_batch", + )?; + if game_creator_agent_runtime_pending_tool_action_exists( + &root, + &runtime.agent_id, + &runtime.run_id, + ) || game_creator_agent_runtime_parallel_read_batch_exists( + &root, + &runtime.agent_id, + &runtime.run_id, + ) { + return Err("Agent Runtime 建立只读并行批次时发现未清理动作账本".to_string()); + } + let durable_runtime = read_game_creator_agent_runtime_for_session_at( + &root, + &runtime.agent_id, + Some(&runtime.session_id), + )? + .state; + if durable_runtime.agent_id != runtime.agent_id + || durable_runtime.task_id != runtime.task_id + || durable_runtime.session_id != runtime.session_id + || durable_runtime.run_id != runtime.run_id + || durable_runtime.source != runtime.source + || durable_runtime.loop_iteration != runtime.loop_iteration + || durable_runtime.applied_steer_cursor != runtime.applied_steer_cursor + { + return Err("Agent Runtime 只读并行批次与当前持久 run 身份不匹配".to_string()); + } + if game_creator_agent_runtime_cancel_requested(&root, &durable_runtime) + || game_creator_agent_runtime_has_queued_steer_after_cursor( + &root, + &runtime.agent_id, + &runtime.run_id, + runtime.applied_steer_cursor, + )? + { + return Ok(AgentRuntimeParallelReadBatchExecution::Stale); + } + if !agent_runtime_parallel_read_batch_is_auto_at(&root, &runtime.agent_id, &actions) { + return Ok(AgentRuntimeParallelReadBatchExecution::NotEligible); + } + let mut pending_actions = Vec::with_capacity(actions.len()); + for (offset, action) in actions.iter().enumerate() { + let mut pending = build_game_creator_agent_runtime_pending_tool_action( + &root, + &durable_runtime, + &task, + &plan, + &observations, + &project_revision_before, + &repository_context_fingerprint, + action, + action_start_index.saturating_add(offset), + AGENT_RUNTIME_ACTION_EXECUTION_MODE_AUTO, + AGENT_RUNTIME_PENDING_ACTION_STATUS_EXECUTING, + None, + )?; + pending.updated_at = unix_timestamp(); + if validate_game_creator_agent_runtime_parallel_read_pending_at_locked( + &root, + &durable_runtime, + &pending, + )? + .is_some() + { + return Ok(AgentRuntimeParallelReadBatchExecution::NotEligible); + } + pending_actions.push(pending); + } + let project_id = game_creator_agent_runtime_context_project_id(&root)?; + let batch_id = agent_runtime_parallel_read_batch_id( + &project_id, + &runtime.agent_id, + &runtime.task_id, + &runtime.session_id, + &runtime.run_id, + runtime.loop_iteration, + &pending_actions, + )?; + let now = unix_timestamp(); + let batch = AgentRuntimeParallelReadBatch { + schema_version: AGENT_RUNTIME_PARALLEL_READ_BATCH_SCHEMA_VERSION.to_string(), + batch_id, + project_id, + agent_id: runtime.agent_id.clone(), + task_id: runtime.task_id.clone(), + session_id: runtime.session_id.clone(), + run_id: runtime.run_id.clone(), + source: runtime.source.clone(), + loop_iteration: runtime.loop_iteration, + planned_steer_cursor: runtime.applied_steer_cursor, + status: AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING.to_string(), + actions: pending_actions, + timings: Vec::new(), + created_at: now, + updated_at: now, + }; + write_game_creator_agent_runtime_parallel_read_batch(&root, &batch)?; + for pending in &batch.actions { + let _ = append_game_creator_agent_runtime_parallel_action_executing_record(&root, pending); + } + execute_game_creator_agent_runtime_parallel_read_members_at_locked(&root, batch) + .map(AgentRuntimeParallelReadBatchExecution::Executed) +} + +async fn execute_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + runtime: &AgentRuntimeState, + task: &str, + plan: &AgentRuntimeToolPlan, + observations: &[AgentRuntimeToolObservation], + project_revision_before: &AgentRuntimeProjectRevision, + repository_context_fingerprint: &str, + action_start_index: usize, + actions: &[AgentRuntimeToolAction], +) -> Result { + let root = root.to_path_buf(); + let runtime = runtime.clone(); + let task = task.to_string(); + let plan = plan.clone(); + let observations = observations.to_vec(); + let project_revision_before = project_revision_before.clone(); + let repository_context_fingerprint = repository_context_fingerprint.to_string(); + let actions = actions.to_vec(); + tokio::task::spawn_blocking(move || { + prepare_and_execute_game_creator_agent_runtime_parallel_read_batch_blocking( + root, + runtime, + task, + plan, + observations, + project_revision_before, + repository_context_fingerprint, + action_start_index, + actions, + ) + }) + .await + .map_err(|error| format!("Agent Runtime 只读并行批次调度失败：{error}"))? +} + +#[cfg(test)] +pub(crate) fn execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + root: &Path, + agent_id: &str, + actions: Vec, +) -> Result<(), String> { + let runtime = read_game_creator_agent_runtime_at(root, agent_id)?.state; + let task = runtime.current_task.clone(); + let plan = AgentRuntimeToolPlan { + thinking_summary: "并行读取测试".to_string(), + plan_update: None, + plan: actions + .iter() + .map(|action| format!("读取 {}", action.tool)) + .collect(), + actions: actions.clone(), + response: String::new(), + }; + let project_revision = read_game_creator_agent_runtime_project_revision(root)?; + let repository_context_fingerprint = build_repository_startup_context_at(root)?.fingerprint; + match prepare_and_execute_game_creator_agent_runtime_parallel_read_batch_blocking( + root.to_path_buf(), + runtime, + task, + plan, + Vec::new(), + project_revision, + repository_context_fingerprint, + 0, + actions, + )? { + AgentRuntimeParallelReadBatchExecution::Executed(_) => Ok(()), + AgentRuntimeParallelReadBatchExecution::NotEligible => { + Err("测试动作未形成只读并行批次".to_string()) + } + AgentRuntimeParallelReadBatchExecution::Stale => { + Err("测试动作在只读并行批次建立前已过期".to_string()) + } + } +} + +#[cfg(test)] +pub(crate) fn project_game_creator_agent_runtime_parallel_read_batch_for_test_at( + root: &Path, + agent_id: &str, +) -> Result { + let mut runtime = read_game_creator_agent_runtime_at(root, agent_id)?.state; + let batch = + read_game_creator_agent_runtime_parallel_read_batch(root, agent_id, &runtime.run_id)?; + let pending = batch + .actions + .first() + .ok_or_else(|| "测试只读并行批次缺少动作".to_string())?; + let task = pending.task.clone(); + let context_bundle = read_game_creator_agent_runtime_context_bundle_with_superseded_goal( + root, + &runtime, + Some(pending), + false, + )?; + let continuation = context_bundle + .map(continuation_from_game_creator_agent_runtime_context_bundle) + .unwrap_or_else(|| AgentRuntimeContinuationContext { + plan: pending.tool_plan(), + observations: pending.observations.clone(), + next_loop_index: usize::try_from(pending.loop_iteration).unwrap_or(usize::MAX), + ..AgentRuntimeContinuationContext::default() + }); + let plan = continuation.plan.clone(); + let mut observations = continuation.observations.clone(); + let mut tracker = AgentRuntimeContextWindowTracker::from_continuation(&continuation); + project_game_creator_agent_runtime_parallel_read_batch( + root, + &mut runtime, + &task, + &plan, + &mut observations, + continuation.next_loop_index, + &mut tracker, + &batch, + )?; + Ok(runtime) +} + +#[cfg(test)] +pub(crate) fn rewind_game_creator_agent_runtime_parallel_read_batch_for_test_at( + root: &Path, + agent_id: &str, +) -> Result, String> { + let runtime = read_game_creator_agent_runtime_at(root, agent_id)?.state; + let mut batch = + read_game_creator_agent_runtime_parallel_read_batch(root, agent_id, &runtime.run_id)?; + let action_ids = batch + .actions + .iter() + .map(|pending| pending.action_id.clone()) + .collect::>(); + for pending in &mut batch.actions { + pending.status = AGENT_RUNTIME_PENDING_ACTION_STATUS_EXECUTING.to_string(); + pending.observation = None; + pending.updated_at = unix_timestamp(); + } + batch.status = AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING.to_string(); + batch.timings.clear(); + batch.updated_at = unix_timestamp(); + write_game_creator_agent_runtime_parallel_read_batch(root, &batch)?; + Ok(action_ids) +} + +#[cfg(test)] +pub(crate) fn recover_game_creator_agent_runtime_parallel_read_batch_for_test_at( + root: &Path, + agent_id: &str, +) -> Result<(), String> { + let runtime = read_game_creator_agent_runtime_at(root, agent_id)?.state; + let batch = + read_game_creator_agent_runtime_parallel_read_batch(root, agent_id, &runtime.run_id)?; + recover_game_creator_agent_runtime_parallel_read_batch_blocking(root.to_path_buf(), batch) + .map(|_| ()) +} + +fn project_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + runtime: &mut AgentRuntimeState, + task: &str, + plan: &AgentRuntimeToolPlan, + observations: &mut Vec, + loop_index: usize, + context_tracker: &mut AgentRuntimeContextWindowTracker, + batch: &AgentRuntimeParallelReadBatch, +) -> Result { + validate_game_creator_agent_runtime_parallel_read_batch(root, batch)?; + if batch.status != AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED { + return Err("Agent Runtime 只读并行批次尚未形成终态 observation".to_string()); + } + let first_pending = batch + .actions + .first() + .ok_or_else(|| "Agent Runtime 只读并行批次缺少动作".to_string())?; + if first_pending.task != task + || runtime.agent_id != batch.agent_id + || runtime.task_id != batch.task_id + || runtime.session_id != batch.session_id + || runtime.run_id != batch.run_id + || runtime.source != batch.source + || runtime.loop_iteration != batch.loop_iteration + { + return Err("Agent Runtime 只读并行批次投影上下文与当前 run 不匹配".to_string()); + } + let stored_batch_start_observations = + sanitize_game_creator_agent_runtime_context_observations_for_storage( + root, + &first_pending.observations, + ); + let stored_observations = + sanitize_game_creator_agent_runtime_context_observations_for_storage(root, observations); + if stored_observations.len() < stored_batch_start_observations.len() + || stored_observations[..stored_batch_start_observations.len()] + != stored_batch_start_observations + { + return Err("Agent Runtime 只读并行批次 context 前缀与批次起点不匹配".to_string()); + } + let context_projected_count = stored_observations.len() - stored_batch_start_observations.len(); + if context_projected_count > batch.actions.len() { + return Err("Agent Runtime 只读并行批次 context 包含批次之外的 observation".to_string()); + } + for (index, pending) in batch + .actions + .iter() + .take(context_projected_count) + .enumerate() + { + let observation = pending + .observation + .as_ref() + .ok_or_else(|| "Agent Runtime 只读并行批次成员缺少 observation".to_string())?; + let stored_observation = + sanitize_game_creator_agent_runtime_context_observations_for_storage( + root, + std::slice::from_ref(observation), + ) + .into_iter() + .next() + .ok_or_else(|| "Agent Runtime 只读并行批次 observation 无法持久化".to_string())?; + if stored_observations[stored_batch_start_observations.len() + index] != stored_observation + { + return Err(format!( + "Agent Runtime 只读并行批次 context observation 顺序或内容冲突：{}", + pending.action_id + )); + } + } + *observations = stored_observations; + + let mut runtime_projected_count = 0_usize; + let mut missing_projection_seen = false; + for pending in &batch.actions { + let observation = pending + .observation + .as_ref() + .ok_or_else(|| "Agent Runtime 只读并行批次成员缺少 observation".to_string())?; + let existing_call = runtime + .recent_tool_calls + .iter() + .find(|call| call.action_id.as_deref() == Some(pending.action_id.as_str())); + if let Some(existing_call) = existing_call { + if missing_projection_seen + || existing_call.tool != observation.tool + || existing_call.status != observation.status + || existing_call.summary != sanitize_agent_runtime_text(&observation.summary, 240) + || existing_call.action_fingerprint.as_deref() + != Some( + agent_runtime_tool_action_fingerprint(&pending.action, &pending.task) + .as_str(), + ) + { + return Err(format!( + "Agent Runtime 只读并行批次已有投影顺序或内容冲突：{}", + pending.action_id + )); + } + runtime_projected_count = runtime_projected_count.saturating_add(1); + } else { + missing_projection_seen = true; + } + } + if context_projected_count > runtime_projected_count + || runtime_projected_count.saturating_sub(context_projected_count) > 1 + { + return Err(format!( + "Agent Runtime 只读并行批次状态与 context 投影前缀不一致：state={runtime_projected_count} context={context_projected_count}" + )); + } + + let mut repository_context_drifted = false; + for (index, pending) in batch.actions.iter().enumerate() { + let observation = pending + .observation + .as_ref() + .ok_or_else(|| "Agent Runtime 只读并行批次成员缺少 observation".to_string())? + .clone(); + let already_projected_to_runtime = index < runtime_projected_count; + let already_projected_to_context = index < context_projected_count; + if !already_projected_to_runtime { + let action_index = usize::try_from(pending.action_index).unwrap_or(usize::MAX); + activate_agent_runtime_plan_step( + runtime, + action_index, + pending + .action + .reason + .as_deref() + .unwrap_or(pending.action.tool.as_str()), + ); + let observation_summary = observation.summary(); + runtime.observations.push(observation_summary.clone()); + append_agent_runtime_tool_call_record( + root, + runtime, + &pending.task, + &pending.action, + &observation, + Some(&pending.action_id), + ); + runtime.pending_tool_action = None; + runtime.status = "running".to_string(); + runtime.phase = "observation".to_string(); + runtime.current_action = format!("并行读取工具 {}", observation.tool); + runtime.waiting_on = "Agent 根据只读工具观察修正计划".to_string(); + runtime.next_step = if observation.is_repository_context_drift() { + "回到同一 run 的下一轮 planning，重新确认适用仓库规范".to_string() + } else { + "按 Provider 顺序整合只读工具观察".to_string() + }; + complete_agent_runtime_active_plan_step( + runtime, + if observation.status == "ok" { + "completed" + } else { + "failed" + }, + &observation_summary, + ); + runtime.updated_at = unix_timestamp(); + let public_observation_detail = + agent_runtime_public_observation_detail(root, &observation); + append_game_creator_agent_runtime_task_projection_once( + root, + runtime, + &pending.action_id, + )?; + refresh_game_creator_agent_runtime_task_queue(root, runtime)?; + write_game_creator_agent_runtime_state(root, runtime)?; + append_game_creator_agent_runtime_action_event( + root, + runtime, + "observation", + runtime.status.as_str(), + "observation", + &observation_summary, + public_observation_detail.as_deref(), + &pending.action_id, + )?; + } else if !already_projected_to_context { + // The runtime state is written before the remaining public projections. Only the + // final projected prefix member can be in this crash window. + append_game_creator_agent_runtime_task_projection_once( + root, + runtime, + &pending.action_id, + )?; + refresh_game_creator_agent_runtime_task_queue(root, runtime)?; + write_game_creator_agent_runtime_state(root, runtime)?; + let observation_summary = observation.summary(); + let public_observation_detail = + agent_runtime_public_observation_detail(root, &observation); + append_game_creator_agent_runtime_action_event( + root, + runtime, + "observation", + runtime.status.as_str(), + "observation", + &observation_summary, + public_observation_detail.as_deref(), + &pending.action_id, + )?; + } + fail_game_creator_agent_runtime_parallel_projection_for_test_at( + root, + "after-public-projection", + index, + )?; + append_agent_runtime_action_receipt( + root, + runtime, + &pending.action_id, + &pending.action_fingerprint, + &observation.tool, + AGENT_RUNTIME_ACTION_EXECUTION_MODE_AUTO, + pending.input_summary.as_deref(), + &observation, + )?; + append_agent_db_terminal_observation_if_missing_for_action( + root, + &runtime.agent_id, + &runtime.run_id, + &pending.action_id, + serde_json::json!({ + "recordType": "agent.runtime.tool_observation", + "agentId": runtime.agent_id, + "taskId": runtime.task_id, + "runId": runtime.run_id, + "tool": observation.tool, + "status": observation.status, + "summary": observation.summary, + "actionId": pending.action_id, + "actionFingerprint": pending.action_fingerprint, + "decision": "auto", + "parallelBatchId": batch.batch_id, + }), + )?; + repository_context_drifted |= observation.is_repository_context_drift(); + if !already_projected_to_context { + context_tracker.record(&observation); + observations.push(observation); + *observations = sanitize_game_creator_agent_runtime_context_observations_for_storage( + root, + observations, + ); + persist_game_creator_agent_runtime_context( + root, + runtime, + task, + plan, + observations, + loop_index, + context_tracker, + )?; + } + fail_game_creator_agent_runtime_parallel_projection_for_test_at( + root, + "after-context", + index, + )?; + } + append_game_creator_agent_runtime_parallel_read_batch_completed_audit(root, batch)?; + let overlap_nanos = agent_runtime_parallel_read_overlap_nanos(&batch.timings); + append_game_creator_agent_runtime_action_event( + root, + runtime, + "parallel_read_batch.completed", + runtime.status.as_str(), + "observation", + &format!( + "Agent 已完成 {} 个只读工具动作并按 Provider 顺序持久化。", + batch.actions.len() + ), + Some(&format!( + "batchId={} · actionCount={} · overlapNanos={} · overlapped={}", + batch.batch_id, + batch.actions.len(), + overlap_nanos, + overlap_nanos > 0 + )), + &first_pending.action_id, + )?; + fail_game_creator_agent_runtime_parallel_projection_for_test_at( + root, + "before-sidecar-remove", + batch.actions.len(), + )?; + remove_game_creator_agent_runtime_parallel_read_batch(root, &batch.agent_id, &batch.run_id)?; + Ok(repository_context_drifted) +} + +#[cfg(test)] +fn fail_game_creator_agent_runtime_parallel_projection_for_test_at( + root: &Path, + stage: &str, + action_index: usize, +) -> Result<(), String> { + let marker = root.join(".agent/runtime/test-parallel-read-projection-failpoint"); + let expected = format!("{stage}:{action_index}"); + if fs::read_to_string(marker) + .ok() + .is_some_and(|value| value.trim() == expected) + { + return Err(format!("测试只读并行投影故障点：{stage}:{action_index}")); + } + Ok(()) +} + +#[cfg(not(test))] +fn fail_game_creator_agent_runtime_parallel_projection_for_test_at( + _root: &Path, + _stage: &str, + _action_index: usize, +) -> Result<(), String> { + Ok(()) +} + fn mark_static_delegate_claim_observed_for_pending_action_at( root: &Path, pending: &AgentRuntimePendingToolAction, @@ -14257,7 +15800,7 @@ pub(crate) async fn compact_game_creator_agent_runtime_session_at( || runtime.task_queue.running > 0 || runtime.task_queue.waiting_for_confirmation > 0 || runtime.task_queue.waiting_for_user_input > 0 - || game_creator_agent_runtime_pending_tool_action_exists( + || game_creator_agent_runtime_has_pending_action_ledger( root, &agent_id, &runtime.state.run_id, @@ -16435,6 +17978,53 @@ fn agent_runtime_pending_reconciliation_observation( } } +pub(crate) fn agent_runtime_tool_is_parallel_safe_read(tool: &str) -> bool { + matches!( + tool.trim(), + "memory.read" + | "conversation.read" + | "asset.list" + | "project.search" + | "project.diff" + | "git.inspect" + | "file.list" + | "file.read" + | "task.list" + ) +} + +pub(crate) fn agent_runtime_parallel_read_batch_len( + actions: &[AgentRuntimeToolAction], + start_index: usize, +) -> usize { + let count = actions + .iter() + .skip(start_index) + .take(AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT) + .take_while(|action| agent_runtime_tool_is_parallel_safe_read(&action.tool)) + .count(); + if count >= 2 { + count + } else { + 0 + } +} + +fn agent_runtime_parallel_read_batch_is_auto_at( + root: &Path, + agent_id: &str, + actions: &[AgentRuntimeToolAction], +) -> bool { + actions.iter().all(|action| { + let Some(command_id) = game_creator_agent_runtime_tool_command_id(action.tool.trim()) + else { + return false; + }; + agent_runtime_tool_is_parallel_safe_read(&action.tool) + && game_creator_agent_runtime_tool_policy_rule(root, agent_id, command_id).is_none() + }) +} + fn game_creator_agent_runtime_tool_command_id(tool: &str) -> Option<&'static str> { match tool { "memory.read" => Some("memory.read"), @@ -16546,6 +18136,256 @@ fn game_creator_agent_runtime_pending_tool_action_relative_path( ) } +fn game_creator_agent_runtime_parallel_read_batch_relative_path( + agent_id: &str, + run_id: &str, +) -> String { + format!( + ".agent/runtime/parallel-read-batches/{}/{}.json", + agent_runtime_confirmation_path_component(agent_id, "agent"), + agent_runtime_confirmation_path_component(run_id, "run") + ) +} + +pub(crate) fn game_creator_agent_runtime_parallel_read_batch_path( + root: &Path, + agent_id: &str, + run_id: &str, +) -> PathBuf { + root.join(game_creator_agent_runtime_parallel_read_batch_relative_path(agent_id, run_id)) +} + +fn game_creator_agent_runtime_parallel_read_batch_exists( + root: &Path, + agent_id: &str, + run_id: &str, +) -> bool { + let path = game_creator_agent_runtime_parallel_read_batch_path(root, agent_id, run_id); + path.exists() || agent_runtime_json_sidecar_backup_path(&path).exists() +} + +fn game_creator_agent_runtime_has_pending_action_ledger( + root: &Path, + agent_id: &str, + run_id: &str, +) -> bool { + game_creator_agent_runtime_pending_tool_action_exists(root, agent_id, run_id) + || game_creator_agent_runtime_parallel_read_batch_exists(root, agent_id, run_id) +} + +fn agent_runtime_parallel_read_batch_id( + project_id: &str, + agent_id: &str, + task_id: &str, + session_id: &str, + run_id: &str, + loop_iteration: u32, + actions: &[AgentRuntimePendingToolAction], +) -> Result { + let action_ids = actions + .iter() + .map(|pending| pending.action_id.as_str()) + .collect::>(); + let identity = serde_json::to_vec(&serde_json::json!({ + "projectId": project_id, + "agentId": agent_id, + "taskId": task_id, + "sessionId": session_id, + "runId": run_id, + "loopIteration": loop_iteration, + "actionIds": action_ids, + })) + .map_err(|error| format!("序列化只读并行批次身份失败：{error}"))?; + let fingerprint = format!("{:x}", Sha256::digest(identity)); + Ok(format!( + "parallel-read-{}", + fingerprint.chars().take(32).collect::() + )) +} + +fn validate_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + batch: &AgentRuntimeParallelReadBatch, +) -> Result<(), String> { + if batch.schema_version != AGENT_RUNTIME_PARALLEL_READ_BATCH_SCHEMA_VERSION { + return Err(format!( + "不支持的 Agent Runtime 只读并行批次版本：{}", + batch.schema_version + )); + } + if batch.project_id != game_creator_agent_runtime_context_project_id(root)? { + return Err("Agent Runtime 只读并行批次项目身份不匹配".to_string()); + } + if !matches!( + batch.status.as_str(), + AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING + | AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED + ) { + return Err(format!( + "Agent Runtime 只读并行批次状态无效：{}", + batch.status + )); + } + if !(2..=AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT).contains(&batch.actions.len()) { + return Err("Agent Runtime 只读并行批次动作数量必须在 2-3 之间".to_string()); + } + if batch.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING + && !batch.timings.is_empty() + { + return Err("执行中的 Agent Runtime 只读并行批次不能提前保存计时结果".to_string()); + } + if batch.status == AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED + && batch.timings.len() != batch.actions.len() + { + return Err("已观察 Agent Runtime 只读并行批次缺少完整计时结果".to_string()); + } + let mut previous_action_index = None; + let mut action_ids = std::collections::BTreeSet::new(); + let first_pending = batch + .actions + .first() + .ok_or_else(|| "Agent Runtime 只读并行批次缺少首个动作".to_string())?; + for (index, pending) in batch.actions.iter().enumerate() { + validate_agent_runtime_pending_tool_action_record(root, pending)?; + if pending.agent_id != batch.agent_id + || pending.task_id != batch.task_id + || pending.session_id != batch.session_id + || pending.run_id != batch.run_id + || pending.source != batch.source + || pending.loop_iteration != batch.loop_iteration + || pending.planned_steer_cursor != batch.planned_steer_cursor + || !pending.is_auto() + || !agent_runtime_tool_is_parallel_safe_read(&pending.action.tool) + { + return Err("Agent Runtime 只读并行批次成员身份或工具分类不匹配".to_string()); + } + if pending.task != first_pending.task + || pending.goal_id != first_pending.goal_id + || pending.goal_revision != first_pending.goal_revision + || pending.goal_snapshot_fingerprint != first_pending.goal_snapshot_fingerprint + || pending.thinking_summary != first_pending.thinking_summary + || pending.plan != first_pending.plan + || pending.fallback_response != first_pending.fallback_response + || pending.observations != first_pending.observations + || pending.project_revision_before != first_pending.project_revision_before + || pending.verification_gate_before != first_pending.verification_gate_before + || pending.planned_repository_context_fingerprint + != first_pending.planned_repository_context_fingerprint + { + return Err("Agent Runtime 只读并行批次成员的 planning 快照不一致".to_string()); + } + if !action_ids.insert(pending.action_id.clone()) { + return Err("Agent Runtime 只读并行批次包含重复 actionId".to_string()); + } + if previous_action_index.is_some_and(|previous| pending.action_index != previous + 1) { + return Err("Agent Runtime 只读并行批次 action index 必须连续递增".to_string()); + } + previous_action_index = Some(pending.action_index); + match batch.status.as_str() { + AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_EXECUTING + if pending.status == AGENT_RUNTIME_PENDING_ACTION_STATUS_EXECUTING + && pending.observation.is_none() => {} + AGENT_RUNTIME_PARALLEL_READ_BATCH_STATUS_OBSERVED + if pending.status == AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED + && pending.observation.as_ref().is_some_and(|observation| { + observation.tool == pending.action.tool + && !observation.is_waiting_for_confirmation() + && !observation.requires_reconciliation() + }) => {} + _ => { + return Err(format!( + "Agent Runtime 只读并行批次成员状态与批次不一致：index={index}" + )); + } + } + } + for (timing, pending) in batch.timings.iter().zip(batch.actions.iter()) { + if timing.action_id != pending.action_id + || timing.started_at_nanos == 0 + || timing.finished_at_nanos < timing.started_at_nanos + { + return Err("Agent Runtime 只读并行批次计时身份或区间无效".to_string()); + } + } + let expected_batch_id = agent_runtime_parallel_read_batch_id( + &batch.project_id, + &batch.agent_id, + &batch.task_id, + &batch.session_id, + &batch.run_id, + batch.loop_iteration, + &batch.actions, + )?; + if batch.batch_id != expected_batch_id { + return Err("Agent Runtime 只读并行批次身份指纹已变化".to_string()); + } + Ok(()) +} + +fn write_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + batch: &AgentRuntimeParallelReadBatch, +) -> Result<(), String> { + validate_game_creator_agent_runtime_parallel_read_batch(root, batch)?; + write_agent_runtime_json_sidecar_with_max_bytes( + root, + &game_creator_agent_runtime_parallel_read_batch_relative_path( + &batch.agent_id, + &batch.run_id, + ), + "Agent Runtime 只读并行批次", + batch, + AGENT_RUNTIME_PARALLEL_READ_BATCH_SIDECAR_MAX_BYTES, + ) +} + +fn read_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + agent_id: &str, + run_id: &str, +) -> Result { + let relative_path = + game_creator_agent_runtime_parallel_read_batch_relative_path(agent_id, run_id); + let batch = read_agent_runtime_json_sidecar_with_max_bytes::( + root, + &relative_path, + "Agent Runtime 只读并行批次", + AGENT_RUNTIME_PARALLEL_READ_BATCH_SIDECAR_MAX_BYTES, + )? + .ok_or_else(|| "Agent Runtime 只读并行批次不存在".to_string())?; + if batch.agent_id != agent_id || batch.run_id != run_id { + return Err("Agent Runtime 只读并行批次 Agent 或 run 身份不匹配".to_string()); + } + validate_game_creator_agent_runtime_parallel_read_batch(root, &batch)?; + Ok(batch) +} + +fn remove_game_creator_agent_runtime_parallel_read_batch( + root: &Path, + agent_id: &str, + run_id: &str, +) -> Result<(), String> { + let path = game_creator_agent_runtime_parallel_read_batch_path(root, agent_id, run_id); + let backup_path = agent_runtime_json_sidecar_backup_path(&path); + remove_agent_runtime_json_sidecar_backup(&backup_path, "Agent Runtime 只读并行批次")?; + match fs::symlink_metadata(&path) { + Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => { + Err("Agent Runtime 只读并行批次必须是普通文件".to_string()) + } + Ok(_) => fs::remove_file(&path).map_err(|error| { + format!( + "删除 Agent Runtime 只读并行批次失败：{}: {error}", + path.display() + ) + }), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!( + "读取 Agent Runtime 只读并行批次元数据失败：{}: {error}", + path.display() + )), + } +} + pub(crate) fn game_creator_agent_runtime_pending_tool_action_exists( root: &Path, agent_id: &str, @@ -16929,7 +18769,8 @@ fn remove_game_creator_agent_runtime_pending_tool_action( "读取 Agent Runtime 待确认动作元数据失败：{}: {error}", path.display() )), - } + }?; + remove_game_creator_agent_runtime_parallel_read_batch(root, agent_id, run_id) } fn remove_game_creator_agent_runtime_confirmations( diff --git a/apps/ai-game-creator-shell/src-tauri/src/tests.rs b/apps/ai-game-creator-shell/src-tauri/src/tests.rs index 8de2f5eec..3c823fe7d 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/tests.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/tests.rs @@ -9,6 +9,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use zip::write::SimpleFileOptions; static TEST_PROJECT_COUNTER: AtomicU64 = AtomicU64::new(0); +static TEST_MOCK_PORT_COUNTER: AtomicU64 = AtomicU64::new(20_000); static TEST_CONFIG_LOCK: StdMutex<()> = StdMutex::new(()); struct TestConfigGuard { @@ -3452,6 +3453,26 @@ fn write_test_canvas_export_zip(path: &Path) { writer.finish().expect("finish canvas export zip"); } +fn bind_test_tcp_listener(label: &str) -> TcpListener { + match TcpListener::bind(("127.0.0.1", 0)) { + Ok(listener) => return listener, + Err(ephemeral_error) => { + for _ in 0..12_000 { + let candidate = + 20_000 + TEST_MOCK_PORT_COUNTER.fetch_add(1, Ordering::Relaxed) % 12_000; + if let Ok(port) = u16::try_from(candidate) { + if let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) { + return listener; + } + } + } + panic!( + "{label}: ephemeral bind failed ({ephemeral_error}); fixed test range exhausted" + ); + } + } +} + fn spawn_mock_llm_server(response_content: String) -> String { spawn_mock_llm_server_responses(vec![response_content]) } @@ -3549,7 +3570,7 @@ fn spawn_mock_llm_server_responses_with_capture( response_contents: Vec, request_sender: Option>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock llm bind"); + let listener = bind_test_tcp_listener("mock llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for response_content in response_contents { @@ -3598,7 +3619,7 @@ fn spawn_interactive_mock_llm_server_with_capture( request_sender: mpsc::Sender, response_receiver: mpsc::Receiver, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("interactive mock llm bind"); + let listener = bind_test_tcp_listener("interactive mock llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for _ in 0..response_count { @@ -3648,7 +3669,7 @@ fn spawn_interruptible_mock_llm_server_with_capture( request_sender: mpsc::Sender, response_receiver: mpsc::Receiver, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("interruptible mock llm bind"); + let listener = bind_test_tcp_listener("interruptible mock llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for _ in 0..response_count { @@ -3684,7 +3705,7 @@ fn spawn_mock_llm_transport_failures_then_response( response_content: String, request_sender: Option>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock transient llm bind"); + let listener = bind_test_tcp_listener("mock transient llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for _ in 0..failure_count { @@ -3726,7 +3747,7 @@ fn spawn_mock_llm_raw_responses_with_capture( response_bodies: Vec, request_sender: Option>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock raw llm bind"); + let listener = bind_test_tcp_listener("mock raw llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for response_body in response_bodies { @@ -3800,7 +3821,7 @@ fn spawn_releasable_mock_llm_raw_response_with_capture( request_sender: mpsc::Sender, release_receiver: mpsc::Receiver<()>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock releasable raw llm bind"); + let listener = bind_test_tcp_listener("mock releasable raw llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { let (mut stream, _) = listener.accept().expect("mock releasable raw llm accept"); @@ -3841,7 +3862,7 @@ fn spawn_releasable_mock_llm_server_responses_with_capture_at( release_index: usize, release_receiver: mpsc::Receiver<()>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock llm bind"); + let listener = bind_test_tcp_listener("mock llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { for (index, response_content) in response_contents.into_iter().enumerate() { @@ -3877,7 +3898,7 @@ fn spawn_mock_llm_stream_server_with_capture( response_body: String, request_sender: Option>, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock stream llm bind"); + let listener = bind_test_tcp_listener("mock stream llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { let (mut stream, _) = listener.accept().expect("mock stream llm accept"); @@ -3925,7 +3946,7 @@ fn spawn_response_stream_mock_llm_server( planning_response: String, final_response: Option, ) -> ResponseStreamMockServer { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("response stream mock bind"); + let listener = bind_test_tcp_listener("response stream mock bind"); let base_url = format!( "http://{}", listener.local_addr().expect("response stream mock addr") @@ -4124,7 +4145,7 @@ fn spawn_mock_llm_stream_fallback_server( mpsc::Sender<()>, std::thread::JoinHandle>, ) { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock fallback llm bind"); + let listener = bind_test_tcp_listener("mock fallback llm bind"); let base_url = format!( "http://{}", listener.local_addr().expect("mock fallback llm addr") @@ -4184,7 +4205,7 @@ fn spawn_barrier_mock_llm_server( expected_requests: usize, request_sender: mpsc::Sender, ) -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock llm bind"); + let listener = bind_test_tcp_listener("mock llm bind"); let base_url = format!("http://{}", listener.local_addr().expect("mock llm addr")); std::thread::spawn(move || { let (mut stream, _) = listener.accept().expect("mock llm accept"); @@ -4233,7 +4254,7 @@ fn spawn_barrier_mock_llm_server( } fn spawn_mock_external_canvas_api_server() -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock canvas api bind"); + let listener = bind_test_tcp_listener("mock canvas api bind"); let base_url = format!( "http://{}", listener.local_addr().expect("mock canvas api addr") @@ -4355,7 +4376,7 @@ fn spawn_mock_external_canvas_api_server() -> String { } fn spawn_mock_external_canvas_generation_failure_server() -> String { - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("mock canvas api bind"); + let listener = bind_test_tcp_listener("mock canvas api bind"); let base_url = format!( "http://{}", listener.local_addr().expect("mock canvas api addr") @@ -6507,6 +6528,762 @@ fn structured_plan_finalization_without_readable_runtime_state_needs_reconciliat } } +#[test] +fn parallel_read_batch_classifier_keeps_effectful_actions_as_ordering_barriers() { + for tool in [ + "memory.read", + "conversation.read", + "asset.list", + "project.search", + "project.diff", + "git.inspect", + "file.list", + "file.read", + "task.list", + ] { + assert!( + agent_runtime_tool_is_parallel_safe_read(tool), + "expected parallel-safe read: {tool}" + ); + } + for tool in [ + "project.index", + "project.verify", + "command.output_read", + "agent.action_history", + "agent.run_status", + "file.write", + "project.git_commit", + "mcp.call", + ] { + assert!( + !agent_runtime_tool_is_parallel_safe_read(tool), + "effectful or claiming tool must stay serial: {tool}" + ); + } + let actions = [ + "file.read", + "project.search", + "file.write", + "file.list", + "task.list", + ] + .into_iter() + .map(|tool| AgentRuntimeToolAction { + tool: tool.to_string(), + reason: Some(format!("测试 {tool}")), + input: serde_json::json!({}), + }) + .collect::>(); + assert_eq!(agent_runtime_parallel_read_batch_len(&actions, 0), 2); + assert_eq!(agent_runtime_parallel_read_batch_len(&actions, 1), 0); + assert_eq!(agent_runtime_parallel_read_batch_len(&actions, 2), 0); + assert_eq!(agent_runtime_parallel_read_batch_len(&actions, 3), 2); +} + +#[test] +fn parallel_read_batch_executes_with_real_overlap_and_projects_provider_order() { + let root = unique_project_path(); + init_local_game_project_at(&root, "project-parallel-read", "并行读取测试项目") + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "ALPHA_PARALLEL_READ").expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "BETA_PARALLEL_READ").expect("write beta fixture"); + let run_id = "parallel-read-overlap-run"; + start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "并行读取两个独立文件", + run_id, + "agent-chat", + "准备并行读取", + vec!["读取 alpha".to_string(), "读取 beta".to_string()], + ) + .expect("start parallel read runtime"); + fs::create_dir_all(root.join(".agent/runtime")).expect("create runtime fixture directory"); + fs::write( + root.join(".agent/runtime/test-parallel-read-barrier-count"), + b"2\n", + ) + .expect("arm parallel read barrier"); + fs::write( + root.join(".agent/runtime/test-parallel-read-delay-first-ms"), + b"80\n", + ) + .expect("delay first provider action"); + let actions = vec![ + AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some("先读取 alpha".to_string()), + input: serde_json::json!({"path": "game/alpha.txt"}), + }, + AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some("再读取 beta".to_string()), + input: serde_json::json!({"path": "game/beta.txt"}), + }, + ]; + execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect("execute parallel read batch"); + + let batch_path = + game_creator_agent_runtime_parallel_read_batch_path(&root, "design-director", run_id); + let batch: Value = + serde_json::from_str(&fs::read_to_string(&batch_path).expect("read parallel read batch")) + .expect("parse parallel read batch"); + assert_eq!(batch["status"], "observed"); + assert_eq!(batch["actions"].as_array().map(Vec::len), Some(2)); + assert_eq!(batch["timings"].as_array().map(Vec::len), Some(2)); + let timings = batch["timings"].as_array().expect("batch timings"); + let latest_start = timings + .iter() + .filter_map(|timing| timing["startedAtNanos"].as_u64()) + .max() + .expect("latest start"); + let earliest_finish = timings + .iter() + .filter_map(|timing| timing["finishedAtNanos"].as_u64()) + .min() + .expect("earliest finish"); + assert!( + latest_start < earliest_finish, + "thread barrier must prove a non-empty physical overlap" + ); + assert!( + timings[1]["finishedAtNanos"].as_u64().unwrap() + < timings[0]["finishedAtNanos"].as_u64().unwrap(), + "second physical read must finish first while projection remains in provider order" + ); + let action_ids = batch["actions"] + .as_array() + .expect("batch actions") + .iter() + .map(|action| action["actionId"].as_str().unwrap().to_string()) + .collect::>(); + + let runtime = project_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + ) + .expect("project parallel read observations"); + assert_eq!(runtime.recent_tool_calls.len(), 2); + assert_eq!( + runtime.recent_tool_calls[0].action_id.as_deref(), + Some(action_ids[0].as_str()) + ); + assert_eq!( + runtime.recent_tool_calls[1].action_id.as_deref(), + Some(action_ids[1].as_str()) + ); + assert!(runtime.recent_tool_calls[0] + .detail + .as_deref() + .is_some_and(|detail| detail.contains("ALPHA_PARALLEL_READ"))); + assert!(runtime.recent_tool_calls[1] + .detail + .as_deref() + .is_some_and(|detail| detail.contains("BETA_PARALLEL_READ"))); + assert!(!batch_path.exists()); + + let records = read_agent_db_records_for_test(&root); + let batch_records = records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.parallel_read_batch.completed" + && record["runId"] == run_id + }) + .collect::>(); + assert_eq!(batch_records.len(), 1); + assert_eq!(batch_records[0]["overlapped"], true); + assert!( + batch_records[0]["overlapNanos"] + .as_u64() + .unwrap_or_default() + > 0 + ); + let receipts = records + .iter() + .filter(|record| { + record["recordType"] == AGENT_RUNTIME_ACTION_RECEIPT_RECORD_TYPE + && record["runId"] == run_id + }) + .map(|record| record["actionId"].as_str().unwrap().to_string()) + .collect::>(); + assert_eq!(receipts, action_ids); + + fs::remove_dir_all(root).ok(); +} + +#[test] +fn parallel_read_batch_respects_confirmation_policy_before_execution() { + let root = unique_project_path(); + init_local_game_project_at(&root, "project-parallel-policy", "并行读取策略测试项目") + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha").expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta").expect("write beta fixture"); + write_project_permission_policy_at( + &root, + ProjectPermissionPolicy { + denied_commands: Vec::new(), + confirm_commands: vec!["file.read".to_string()], + agent_policies: BTreeMap::new(), + }, + ) + .expect("write confirmation policy"); + let run_id = "parallel-read-policy-run"; + start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "读取两个文件", + run_id, + "agent-chat", + "准备读取", + vec!["读取文件".to_string()], + ) + .expect("start policy runtime"); + let actions = ["game/alpha.txt", "game/beta.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + let error = execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect_err("confirmation policy must prevent parallel auto execution"); + assert!(error.contains("未形成只读并行批次")); + assert!( + !game_creator_agent_runtime_parallel_read_batch_path(&root, "design-director", run_id,) + .exists() + ); + fs::remove_dir_all(root).ok(); +} + +#[test] +fn parallel_read_batch_executing_recovery_reuses_action_ids_without_duplicate_audit() { + let root = unique_project_path(); + init_local_game_project_at(&root, "project-parallel-recovery", "并行读取恢复测试项目") + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha recovery").expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta recovery").expect("write beta fixture"); + fs::write(root.join("game/gamma.txt"), "gamma recovery").expect("write gamma fixture"); + let run_id = "parallel-read-recovery-run"; + start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "恢复两个只读动作", + run_id, + "agent-chat", + "准备读取", + vec!["读取文件".to_string()], + ) + .expect("start recovery runtime"); + let actions = ["game/alpha.txt", "game/beta.txt", "game/gamma.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect("execute initial batch"); + let action_ids = + rewind_game_creator_agent_runtime_parallel_read_batch_for_test_at(&root, "design-director") + .expect("rewind batch to executing crash window"); + recover_game_creator_agent_runtime_parallel_read_batch_for_test_at(&root, "design-director") + .expect("recover executing read batch"); + let batch: Value = serde_json::from_str( + &fs::read_to_string(game_creator_agent_runtime_parallel_read_batch_path( + &root, + "design-director", + run_id, + )) + .expect("read recovered batch"), + ) + .expect("parse recovered batch"); + assert_eq!(batch["status"], "observed"); + assert_eq!( + batch["actions"] + .as_array() + .unwrap() + .iter() + .map(|action| action["actionId"].as_str().unwrap().to_string()) + .collect::>(), + action_ids + ); + project_game_creator_agent_runtime_parallel_read_batch_for_test_at(&root, "design-director") + .expect("project recovered batch"); + let records = read_agent_db_records_for_test(&root); + for record_type in [ + "agent.runtime.tool_action.executing", + "agent.runtime.tool_action.observed", + AGENT_RUNTIME_ACTION_RECEIPT_RECORD_TYPE, + ] { + let ids = records + .iter() + .filter(|record| record["recordType"] == record_type && record["runId"] == run_id) + .map(|record| record["actionId"].as_str().unwrap().to_string()) + .collect::>(); + assert_eq!(ids, action_ids, "duplicate or reordered {record_type}"); + } + assert_eq!( + records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.parallel_read_batch.completed" + && record["runId"] == run_id + }) + .count(), + 1 + ); + fs::remove_dir_all(root).ok(); +} + +#[test] +fn parallel_read_batch_projection_recovers_crash_windows_without_duplicate_artifacts() { + for failpoint in [ + None, + Some("after-public-projection:0"), + Some("after-context:0"), + Some("before-sidecar-remove:2"), + ] { + let root = unique_project_path(); + let suffix = failpoint.unwrap_or("observed-sidecar").replace(':', "-"); + let run_id = format!("parallel-read-projection-recovery-{suffix}"); + init_local_game_project_at( + &root, + &format!("project-{suffix}"), + "并行读取投影恢复测试项目", + ) + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha projection recovery") + .expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta projection recovery") + .expect("write beta fixture"); + start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "恢复只读批次公共投影", + &run_id, + "agent-chat", + "准备读取", + vec!["读取两个文件".to_string()], + ) + .expect("start projection recovery runtime"); + let actions = ["game/alpha.txt", "game/beta.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect("execute observed parallel batch"); + let batch_path = + game_creator_agent_runtime_parallel_read_batch_path(&root, "design-director", &run_id); + let batch: Value = serde_json::from_str( + &fs::read_to_string(&batch_path).expect("read observed batch before projection"), + ) + .expect("parse observed batch before projection"); + let action_ids = batch["actions"] + .as_array() + .expect("batch actions") + .iter() + .map(|action| action["actionId"].as_str().unwrap().to_string()) + .collect::>(); + + if let Some(failpoint) = failpoint { + let marker = root.join(".agent/runtime/test-parallel-read-projection-failpoint"); + fs::write(&marker, failpoint).expect("arm projection failpoint"); + let error = project_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + ) + .expect_err("projection failpoint must preserve the observed sidecar"); + assert!(error.contains("测试只读并行投影故障点")); + assert!(batch_path.exists()); + fs::remove_file(marker).expect("disarm projection failpoint"); + } + + let runtime = project_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + ) + .expect("resume observed parallel projection"); + assert!(!batch_path.exists()); + let projected_action_ids = runtime + .recent_tool_calls + .iter() + .filter_map(|call| call.action_id.clone()) + .filter(|action_id| action_ids.contains(action_id)) + .collect::>(); + assert_eq!(projected_action_ids, action_ids); + + let records = read_agent_db_records_for_test(&root); + for record_type in [ + "agent.runtime.tool_action.executing", + "agent.runtime.tool_action.observed", + AGENT_RUNTIME_ACTION_RECEIPT_RECORD_TYPE, + "agent.runtime.tool_observation", + ] { + let ids = records + .iter() + .filter(|record| record["recordType"] == record_type && record["runId"] == run_id) + .map(|record| record["actionId"].as_str().unwrap().to_string()) + .collect::>(); + assert_eq!(ids, action_ids, "duplicate or reordered {record_type}"); + } + assert_eq!( + records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.parallel_read_batch.completed" + && record["runId"] == run_id + }) + .count(), + 1 + ); + let events = fs::read_to_string(game_creator_agent_runtime_event_path( + &root, + "design-director", + )) + .expect("read projection events") + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str::(line).expect("parse projection event")) + .collect::>(); + for action_id in &action_ids { + assert_eq!( + events + .iter() + .filter(|event| { + event["runId"] == run_id + && event["eventType"] == "observation" + && event["actionId"] == action_id.as_str() + }) + .count(), + 1, + "observation event must be unique for {action_id}" + ); + } + assert_eq!( + events + .iter() + .filter(|event| { + event["runId"] == run_id + && event["eventType"] == "parallel_read_batch.completed" + }) + .count(), + 1 + ); + let context = read_game_creator_agent_runtime_context_bundle(&root, &runtime) + .expect("read recovered context") + .expect("recovered context exists"); + assert_eq!(context.observations.len(), 2); + + fs::remove_dir_all(root).ok(); + } +} + +#[test] +fn parallel_read_batch_recovery_blocks_replay_after_cancel_goal_or_repository_drift() { + for stale_kind in ["cancel", "goal", "repository"] { + let root = unique_project_path(); + let run_id = format!("parallel-read-stale-{stale_kind}-run"); + init_local_game_project_at( + &root, + &format!("project-parallel-stale-{stale_kind}"), + "并行读取过期恢复测试项目", + ) + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha before stale recovery") + .expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta before stale recovery") + .expect("write beta fixture"); + let mut state = start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "恢复前检查控制和一致性快照", + &run_id, + "agent-chat", + "准备读取", + vec!["读取两个文件".to_string()], + ) + .expect("start stale recovery runtime"); + if stale_kind == "goal" { + seed_game_creator_agent_goal_for_runtime_test_at( + &root, + &mut state, + "完成第一版并行读取目标", + AGENT_GOAL_STATUS_ACTIVE, + ) + .expect("bind active Goal before batch"); + } + let actions = ["game/alpha.txt", "game/beta.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect("execute initial stale recovery batch"); + rewind_game_creator_agent_runtime_parallel_read_batch_for_test_at(&root, "design-director") + .expect("rewind batch to executing state"); + + match stale_kind { + "cancel" => write_game_creator_agent_runtime_cancel_request( + &root, + "design-director", + &run_id, + "测试恢复前取消", + ) + .expect("write durable cancel request"), + "goal" => { + let mut durable = read_game_creator_agent_runtime_at(&root, "design-director") + .expect("read Goal runtime before revision") + .state; + revise_game_creator_agent_goal_for_runtime_test_at( + &root, + &mut durable, + "完成第二版并行读取目标", + ) + .expect("revise Goal before recovery"); + } + "repository" => fs::write( + root.join("AGENTS.md"), + "# Repository rules changed before recovery\n", + ) + .expect("change repository startup context"), + _ => unreachable!(), + } + + recover_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + ) + .expect("stale executing batch becomes blocked observations"); + let batch: Value = serde_json::from_str( + &fs::read_to_string(game_creator_agent_runtime_parallel_read_batch_path( + &root, + "design-director", + &run_id, + )) + .expect("read blocked recovered batch"), + ) + .expect("parse blocked recovered batch"); + assert_eq!(batch["status"], "observed"); + assert!(batch["actions"].as_array().unwrap().iter().all(|action| { + action["status"] == AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED + && action["observation"]["status"] == "blocked" + && action["observation"]["summary"] + .as_str() + .is_some_and(|summary| summary.contains("旧只读批次未重放")) + })); + + fs::remove_dir_all(root).ok(); + } +} + +#[test] +fn parallel_read_batch_does_not_start_after_steer_is_accepted() { + let root = unique_project_path(); + init_local_game_project_at( + &root, + "project-parallel-steer-first", + "并行读取先 steer 测试", + ) + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha").expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta").expect("write beta fixture"); + let run_id = "parallel-read-steer-first-run"; + let state = start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "读取两个文件", + run_id, + "agent-chat", + "准备读取", + vec!["读取文件".to_string()], + ) + .expect("start steer-first runtime"); + let steer = steer_game_creator_agent_runtime_task_at( + &root, + "design-director", + &state.session_id, + run_id, + "parallel-read-steer-first", + "先不要读取，重新确认目标", + "test", + ) + .expect("accept steer before read batch"); + assert_eq!(steer.status, "queued"); + let actions = ["game/alpha.txt", "game/beta.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + let error = execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &root, + "design-director", + actions, + ) + .expect_err("accepted steer must stale the old read batch"); + assert!(error.contains("建立前已过期")); + assert!( + !game_creator_agent_runtime_parallel_read_batch_path(&root, "design-director", run_id,) + .exists() + ); + assert!(!read_agent_db_records_for_test(&root).iter().any(|record| { + record["recordType"] == "agent.runtime.tool_action.executing" && record["runId"] == run_id + })); + fs::remove_dir_all(root).ok(); +} + +#[test] +fn parallel_read_batch_finishes_before_concurrent_steer_is_accepted() { + let root = unique_project_path(); + init_local_game_project_at( + &root, + "project-parallel-steer-after", + "并行读取后 steer 测试", + ) + .expect("project init"); + fs::write(root.join("game/alpha.txt"), "alpha held read").expect("write alpha fixture"); + fs::write(root.join("game/beta.txt"), "beta held read").expect("write beta fixture"); + let run_id = "parallel-read-steer-after-run"; + let state = start_game_creator_agent_runtime_task_at( + &root, + "design-director", + "并行读取时接收 steer", + run_id, + "agent-chat", + "准备读取", + vec!["读取文件".to_string()], + ) + .expect("start steer-after runtime"); + fs::create_dir_all(root.join(".agent/runtime")).expect("create runtime fixture directory"); + fs::write( + root.join(".agent/runtime/test-parallel-read-barrier-count"), + b"2\n", + ) + .expect("arm parallel barrier"); + fs::write( + root.join(".agent/runtime/test-parallel-read-hold"), + b"hold\n", + ) + .expect("arm parallel hold"); + let actions = ["game/alpha.txt", "game/beta.txt"] + .into_iter() + .map(|path| AgentRuntimeToolAction { + tool: "file.read".to_string(), + reason: Some(format!("读取 {path}")), + input: serde_json::json!({"path": path}), + }) + .collect::>(); + let batch_root = root.clone(); + let batch_thread = std::thread::spawn(move || { + execute_game_creator_agent_runtime_parallel_read_batch_for_test_at( + &batch_root, + "design-director", + actions, + ) + }); + let batch_path = + game_creator_agent_runtime_parallel_read_batch_path(&root, "design-director", run_id); + for _ in 0..200 { + if batch_path.exists() { + break; + } + std::thread::sleep(Duration::from_millis(5)); + } + assert!( + batch_path.exists(), + "executing batch sidecar must become durable" + ); + + let (steer_sender, steer_receiver) = mpsc::channel(); + let steer_root = root.clone(); + let steer_session_id = state.session_id.clone(); + let steer_thread = std::thread::spawn(move || { + let result = steer_game_creator_agent_runtime_task_at( + &steer_root, + "design-director", + &steer_session_id, + run_id, + "parallel-read-steer-after", + "读取结束后按新要求继续", + "test", + ); + steer_sender.send(result).expect("send steer result"); + }); + assert!( + steer_receiver + .recv_timeout(Duration::from_millis(60)) + .is_err(), + "steer acceptance must wait behind the physical read batch boundary" + ); + fs::write( + root.join(".agent/runtime/test-parallel-read-release"), + b"release\n", + ) + .expect("release parallel reads"); + batch_thread + .join() + .expect("join batch thread") + .expect("parallel batch completion"); + let steer = steer_receiver + .recv_timeout(Duration::from_secs(2)) + .expect("steer result after batch") + .expect("accept steer after batch"); + steer_thread.join().expect("join steer thread"); + assert_eq!(steer.status, "queued"); + let batch: Value = + serde_json::from_str(&fs::read_to_string(&batch_path).expect("read observed batch")) + .expect("parse observed batch"); + assert_eq!(batch["status"], "observed"); + assert!(batch["actions"].as_array().unwrap().iter().all(|action| { + action["observation"]["status"] == "ok" + && action["status"] == AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED + })); + project_game_creator_agent_runtime_parallel_read_batch_for_test_at(&root, "design-director") + .expect("project batch completed before steer"); + assert_eq!( + read_game_creator_agent_runtime_at(&root, "design-director") + .expect("read runtime") + .state + .applied_steer_cursor, + 0, + "test projection does not consume the queued steer" + ); + fs::remove_dir_all(root).ok(); +} + #[tokio::test] async fn background_agent_runtime_task_executes_plan_tool_observation_loop() { let root = unique_project_path(); @@ -25054,7 +25831,7 @@ async fn background_provider_failure_redacts_all_persisted_runtime_surfaces() { "request failed url={provider_url_sentinel} path={} task={provider_task_sentinel} secret={provider_secret_sentinel}", provider_path_sentinel.display() ); - let listener = TcpListener::bind(("127.0.0.1", 0)).expect("provider error mock bind"); + let listener = bind_test_tcp_listener("provider error mock bind"); let base_url = format!( "http://{}", listener.local_addr().expect("provider error mock addr") diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 1ae76be13..174619535 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -16,6 +16,16 @@ --- +## 2026-07-16 AI 游戏创作 Agent Runtime 只并行持久只读批次 + +- 背景：V1.26 已允许 Provider 一轮返回最多三个原生工具 action，但同一 Agent 仍逐个执行；直接把 action future `join` 会因同步文件 I/O、单 pending sidecar 和项目一致性锁而形成假并行，并破坏 steer、崩溃恢复与 exactly-once。 +- 决策：同一 Agent 只把连续 2-3 个自动批准的严格只读工具组成 durable parallel-read batch。首版白名单为 `memory.read`、`conversation.read`、`asset.list`、`project.search`、`project.diff`、`git.inspect`、`file.list`、`file.read`、`task.list`；所有写入、命令/进程、确认、Provider/MCP、生成、Git commit、委派和回执认领工具继续串行。批次在项目一致性锁内完成 preflight、executing、线程并行读取和 observed 落盘，控制请求只在批次边界前或后线性化；终态观察仍按 Provider 顺序投影。 +- 恢复与安全：批次成员使用稳定 actionId/指纹。Runner 在 `executing` 中退出只允许同身份重放严格只读物理读取，`observed` 只补齐幂等投影；不能新建 Provider lifecycle、action 或 receipt。公共审计只记录身份、工具名、计时与重叠结论，不记录参数、观察正文、绝对路径或凭据。任何分类、策略、Goal、steer、repository context 或持久身份不确定都失败关闭或退回既有串行路径。 +- 影响范围：AI 游戏创作客户端 Rust Agent Runtime、Runner 恢复、定向测试、真实 Provider E2E 和 Runtime 技术文档。 +- 验证方式：`parallel_read_batch_` 的 8 项确定性用例覆盖真实重叠、稳定顺序、串行屏障、控制竞态、取消/Goal/repository drift、`executing` 重放和 `observed`/部分投影恢复去重；正式 `openai_chat / gpt-5.5` 独立 suite 必须证明模型自主发出至少两个同轮读取、时间区间真实重叠、同 run 唯一回复、原生协议、零重放/重复/泄漏和隔离清理。 +- 真实结论：2026-07-16 隔离 `parallel-read` suite PASS。真实模型同轮提交 2 个独立 `project.search`，单一持久批次重叠 `10,969,247ns` 且按 Provider 顺序投影；4/4 个成功工具计划与 2/2 个 repair 全为 `native_runtime_tools`，6 个 tool-plan 与 1 个 final-reply lifecycle 唯一闭合。最终 assistant/completed 各 1，重复 action/receipt/Provider lifecycle、遗留 finalization/批次 sidecar、私有正文、API Key、诱饵、项目/配置路径和报告泄漏均为 0，隔离 Runner/AppData/项目完整清理。V1.27 当前门禁为 PASS。 +- 关联文档：`docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md`。 + ## 2026-07-16 AI 游戏创作 Goal 真实验收强制原生工具协议 - 背景：V1.18 的 `goal-runtime` 已证明 edit/pause/Runner 强杀/resume/finalization，但该 PASS 早于 V1.26 原生工具目录；旧验收只接受协议兼容集合，无法证明长任务没有静默退回 wrapper/text JSON。阶段等待在 Runtime 已因外部 transport 失败时还可能继续等 30 分钟。 diff --git a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md index 368f97350..3a951a1be 100644 --- a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md +++ b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md @@ -1004,6 +1004,19 @@ V1.26 把 OpenAI-compatible planning 从单个 `submit_agent_tool_plan` 包装 2026-07-16 正式 `openai_chat / gpt-5.5` 的 `project-skill` suite **PASS**。首轮真实执行在匹配 Skill 读取后暴露旧 parser 拒绝 plan-only，保留现场复验进一步证明模型会先单独调用 `update_agent_plan`；Runtime 空动作分支原本已能持久化计划并安全进入下一轮，因此移除矛盾的 parser 拒绝并补三轮确定性闭环。最终加强门禁复跑记录 31 条 task、57 条 event、94 条 Agent DB、6 个成功工具动作和 2 个确认动作；9/9 个成功工具计划与 6/6 个格式修复全部使用 `native_runtime_tools`，旧 wrapper 与 text JSON fallback 均为 0。Agent 在首个项目修改前读取匹配 Skill 1 次、无关 Skill 0 次，只修改 1 个目标文件，Agent `project.verify` 与宿主复验均通过；15 个 tool-plan 加 1 个 final-reply Provider lifecycle 全部唯一闭合，最终 assistant/completed 各 1，重复 message/receipt、遗留 finalization、Skill 正文、API Key、诱饵、项目/正式配置路径和报告泄漏均为 0，隔离 Runner、AppData 与一次性项目完整清理。确定性 Tauri 全量为 822 passed / 4 ignored；V1.26 真实行为门禁至此完成。 +## V1.27 同 Agent 持久只读并行批次 + +V1.27 在 V1.26 一次最多三个原生 action 的基础上，让同一 Agent 可以真实重叠执行彼此独立的本地只读工具。并行不是通用 action 调度器，也不改变不同 Agent 已有的 lane 并行；Runtime 只把连续 2-3 个、当前策略为自动批准且列入严格白名单的读取动作组成一个持久批次，其余动作继续按 Provider 顺序串行。 + +- 首版并行白名单固定为 `memory.read`、`conversation.read`、`asset.list`、`project.search`、`project.diff`、`git.inspect`、`file.list`、`file.read` 和 `task.list`。`project.index` 会刷新持久启动上下文，不属于纯读取；`command.output_read`、`agent.action_history` 和 `agent.run_status` 可能产生审计、屏障或回执认领，也不进入批次。写入、确认、命令/进程、Git commit、验证、预览、图片/生成 Provider、MCP、消息、委派、动态 child 和其它外部副作用全部保持串行。 +- 只合并 Provider 顺序中连续的安全读取；任一非安全动作形成顺序屏障。例如 `read A / read B / write C / read D` 只并行前两个读取。策略要求确认或拒绝、工具目录变化、Goal/steer/repository context 身份漂移时不建立批次，不能把确认消费、策略错误或旧动作暗中转成自动读取。 +- 私有事实源新增按 `project / agent / task / session / run / loop` 绑定的 durable parallel-read batch，最多保存三个稳定 action identity、输入指纹、Provider 顺序、执行状态、终态 observation 和无正文计时元数据。状态只允许 `executing -> observed`；Runner 在 `executing` 中退出时，因为成员工具已由严格分类证明无副作用，可以在同一 action identity 下重新读取，不能创建新 action、receipt 或 Provider request。`observed` 恢复只补齐缺失投影，不重新执行物理读取。 +- 批次执行在一个项目一致性锁内完成预检、`executing` 落盘、真实工作线程并行读取和 `observed` 原子落盘。steer、Goal edit/pause 和 Runtime 写动作必须在该锁之后才被接受，因此整个批次是一个明确线性化边界：控制请求先获得锁则旧批次零执行，批次先获得锁则其全部读取先完成，再接受控制请求。取消仍遵循“已进入工具的动作返回后停止”，不强杀线程。 +- 物理完成顺序不能影响语义。Runtime 始终按 Provider action index 依次写 task/event/receipt/Agent DB、更新结构化计划和 context bundle；每个 actionId 的 terminal observation 与 receipt 必须唯一。公共并行审计只保存 batch/action identity、工具名、数量、时间区间和是否存在真实重叠，不保存 input、observation 正文、项目绝对路径或凭据。 +- 确定性验收必须覆盖白名单/拒绝清单、连续分组与串行屏障、两个及三个读取的真实线程重叠、完成顺序反转但观察顺序稳定、策略变化、steer 先后双向竞态、取消、Goal/repository drift、`executing` 重放、`observed` 只补投影、Runner 强杀、重复 receipt/event/Agent DB 为 0 以及公共零正文。真实 Provider 必须在未提供工具名和顺序配方的任务中自主同轮发出至少两个独立读取，审计时间区间证明重叠，随后同一 Agent/Session/run 完成唯一回复，并证明原生工具协议、稳定 action 顺序、零副作用重放、零重复、零泄漏和隔离现场清理。 + +2026-07-16 正式 `openai_chat / gpt-5.5` 的隔离 `parallel-read` suite **PASS**。一次性项目构造 421 个只读语料文件，真实模型在同一个原生 planning 轮次提交 2 个独立 `project.search`，Runtime 只形成 1 个持久并行批次；两个物理搜索重叠 `10,969,247ns`，公共投影与 receipt 均保持 Provider action 顺序。最终记录 13 条 task、23 条 event 和 48 条 Agent DB；4/4 个成功工具计划与 2/2 个格式修复都使用 `native_runtime_tools`，wrapper/text JSON fallback 为 0，6 个 tool-plan 和 1 个 final-reply Provider lifecycle 全部唯一闭合。最终 assistant/completed 各 1，重复 action lifecycle、receipt、Provider lifecycle、遗留 finalization/批次 sidecar，以及仓库私有正文、API Key、诱饵、项目/正式配置绝对路径和报告泄漏均为 0；隔离 Runner、AppData 与 disposable 项目完整清理。V1.27 真实行为门禁至此完成。 + ## 验收命令 - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml structured_plan_ -- --nocapture` @@ -1011,6 +1024,7 @@ V1.26 把 OpenAI-compatible planning 从单个 `submit_agent_tool_plan` 包装 - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml goal_context_bundle_v4_migrates_v3_and_v2_then_rejects_plan_mismatch -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml response_stream_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml mcp_ -- --nocapture` +- `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml parallel_read_batch_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml swarm_cli::tests -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml typed_goal_pause_and_cancel_require_durable_intent_and_keep_exact_run -- --nocapture` - `npm run ai-game-creator-shell:typecheck` @@ -1028,6 +1042,7 @@ V1.26 把 OpenAI-compatible planning 从单个 `submit_agent_tool_plan` 包装 - `npm run ai-game-creator-shell:agent-runtime:real-e2e -- --config-dir --suite user-input-runtime` - `npm run ai-game-creator-shell:agent-runtime:real-e2e -- --config-dir --suite scoped-agents` - `npm run ai-game-creator-shell:agent-runtime:real-e2e -- --config-dir --suite project-skill` +- `npm run ai-game-creator-shell:agent-runtime:real-e2e -- --config-dir --suite parallel-read` - `npm run ai-game-creator-shell:agent-runtime:real-e2e -- --config-dir --suite full` - `npm run check:encoding` - `git diff --check` diff --git a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md index 78b6f4beb..c1f682dec 100644 --- a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md +++ b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md @@ -577,4 +577,6 @@ game-project/ - 2026-07-16 起，同一 Runtime 文档的“V1.26 Provider 原生工具目录”作为 OpenAI-compatible planning 协议事实源。Chat / Responses 不再只广告 `submit_agent_tool_plan` 包装函数，而是直接提供 `update_agent_plan`、`respond_to_user`、全部内置 Runtime action 和动态 MCP function；每个函数使用独立 schema，Runtime 继续负责身份、权限、确认、沙箱、revision、验证、恢复与副作用防重放。Anthropic 与历史 fixture 保留 text JSON / wrapper 解析兼容，但新请求和 repair 不能静默降级。plan-only 是合法持久 checkpoint，未完成计划仍阻止最终化。 - 2026-07-16 V1.26 已完成真实验收：正式 `openai_chat / gpt-5.5` 的 `project-skill` suite 中 9/9 个成功工具计划与 6/6 个格式修复全部使用 `native_runtime_tools`，wrapper/text fallback 均为 0。Agent 自主读取匹配 Skill、只改唯一目标文件并完成 Agent/宿主双重验证；15 个 tool-plan 和 1 个 final-reply lifecycle 唯一闭合，最终 assistant/completed 各 1，重复、Skill 正文、API Key、诱饵、项目/配置路径和报告泄漏均为 0，隔离 Runner/AppData/项目完整清理。 - 2026-07-16 V1.26 后重新加强并复验 `goal-runtime`：Goal suite 现在把成功计划、repair、call metadata、wrapper/text fallback 和协议审计零 payload 纳入硬门禁。正式 `openai_chat / gpt-5.5` 最终复跑的成功计划 21/21、repair 17/17 全为 `native_runtime_tools`；Goal edit、旧动作失效、真实失败修复、pause、Runner 强杀、显式同 run resume、verification、finalization 和唯一回复全部 PASS，重复、重放、正文、密钥、诱饵与路径泄漏均为 0。Goal 阶段等待同时增加 terminal fail-fast，Provider transport failure 不再占满 30 分钟验收超时。 +- 2026-07-16 起，V1.27 只允许同一 Agent 把连续 2-3 个自动批准的严格只读 action 组成持久并行批次。首版白名单为 `memory.read / conversation.read / asset.list / project.search / project.diff / git.inspect / file.list / file.read / task.list`；写入、确认、命令/进程、验证、生成、MCP、Git commit、委派和回执认领仍按 Provider 顺序串行。批次用项目一致性锁形成 steer/Goal/取消的线性化边界，物理读取由独立工作线程重叠执行，observation/task/event/receipt/context 仍按 Provider action index 稳定投影；`executing` Runner 恢复只重放严格只读观察，`observed` 只补投影。 +- V1.27 的 8 项确定性回归已证明：2 个读取真实非空重叠；第二个物理读取先结束时仍按 Provider 顺序投影；3 个动作 `executing` 恢复保持原 actionId；`observed` sidecar、部分公共投影、部分 context 和删除 sidecar 前故障都从持久前缀继续且零重复；确认策略阻止自动批次；取消、Goal/repository drift 阻止旧读取重放；steer 先获得锁时旧批次零执行，批次先获得锁时 steer 等到全部读取 observed 后才接受。正式 `openai_chat / gpt-5.5` 的隔离 `parallel-read` suite 同样 PASS：真实模型同轮提交 2 个独立 `project.search`，物理重叠 `10,969,247ns`，4/4 个成功工具计划与 2/2 个 repair 全为 `native_runtime_tools`，6 个 tool-plan 与 1 个 final-reply lifecycle 唯一闭合，最终 assistant/completed 各 1；重复、正文/API Key/诱饵/项目与配置路径泄漏均为 0，Runner/AppData/项目完整清理。V1.27 当前真实行为门禁为 PASS。 - 开发模式可通过本地项目文件面板执行 `file.list/read/write/delete`，普通用户界面不暴露文件面板。