diff --git a/apps/api/src/lib/__tests__/memory-topics.test.ts b/apps/api/src/lib/__tests__/memory-topics.test.ts index 87e31a03..1e5dbc96 100644 --- a/apps/api/src/lib/__tests__/memory-topics.test.ts +++ b/apps/api/src/lib/__tests__/memory-topics.test.ts @@ -14,9 +14,6 @@ vi.mock('../logger.js', () => ({ import { env } from '../../env.js' import { readMemoryFile, writeMemoryFile } from '../memory-storage.js' import { - MEMORY_MAIN_FILE, - MEMORY_TOPIC_HARD_TOKENS, - MemoryTopicError, applyInsightToTopics, archiveMemoryTopic, detectMemoryHierarchyMode, @@ -24,6 +21,10 @@ import { getValidatedMemoryMain, hashMemoryBlock, listMemoryTopics, + MEMORY_MAIN_FILE, + MEMORY_TOPIC_HARD_TOKENS, + MemoryTopicError, + mergeMemoryTopics, parseMemoryTopicFile, reactivateMemoryTopic, readMemoryTopic, @@ -33,6 +34,7 @@ import { replaceTopicBody, selectMemoryTopicForRecall, splitMemoryTopic, + topicPath, } from '../memory-topics.js' let testRoot: string @@ -490,6 +492,60 @@ describe('memory-topics', () => { expect(readMemoryFile('agt_test', MEMORY_MAIN_FILE)).toContain(`\`${topicId}\``) }) + it('restores the merge target when archiving a source fails midway', async () => { + // Distinct scopes/keywords so each insight opens its own topic instead of + // being merged into the previous one; two items each clears the + // "insufficient_new_topic_content" guard on new-topic creation. + const target = applyInsightToTopics('agt_test', insight()) + const sourceA = applyInsightToTopics( + 'agt_test', + insight({ + title: 'Billing invoices', + scope: 'Billing invoice lifecycle.', + description: 'Invoice rules.', + keywords: ['billing', 'invoice'], + items: ['Invoices settle nightly.', 'Refunds post next day.'], + }), + ) + const sourceB = applyInsightToTopics( + 'agt_test', + insight({ + title: 'Render pipeline', + scope: 'GPU render pipeline stages.', + description: 'Render rules.', + keywords: ['render', 'gpu'], + items: ['Shaders compile at boot.', 'Frame budget is 16ms.'], + }), + ) + const targetTopicId = target.topic?.topicId as string + const sourceAId = sourceA.topic?.topicId as string + const sourceBId = sourceB.topic?.topicId as string + expect(targetTopicId && sourceAId && sourceBId).toBeTruthy() + + const targetBefore = readMemoryTopic('agt_test', targetTopicId).body + const mainBefore = readMemoryFile('agt_test', MEMORY_MAIN_FILE) + + // mergeMemoryTopics reads every source up front, writes the target, then + // archives the sources in a loop. Occupy source B's archive destination + // with a *directory*, so its archive write throws EISDIR after source A + // already archived cleanly — the same shape as a mid-loop ENOSPC/EIO. + const sourceBArchivePath = topicPath({ + topicId: sourceBId, + title: 'Render pipeline', + status: 'archived', + }) + mkdirSync(join(testRoot, 'agt_test', sourceBArchivePath), { recursive: true }) + + expect(() => mergeMemoryTopics('agt_test', [sourceAId, sourceBId], targetTopicId)).toThrow() + + // The target must not be left holding merged content while a source is + // still active — that duplicates facts and makes the retry unrecoverable. + expect(readMemoryTopic('agt_test', targetTopicId).body).toBe(targetBefore) + expect(readMemoryTopic('agt_test', sourceAId).status).toBe('active') + expect(readMemoryTopic('agt_test', sourceBId).status).toBe('active') + expect(readMemoryFile('agt_test', MEMORY_MAIN_FILE)).toBe(mainBefore) + }) + it('splits a topic only when every source block is copied verbatim exactly once', async () => { const created = applyInsightToTopics('agt_test', insight()) const sourceTopicId = created.topic?.topicId as string diff --git a/apps/api/src/lib/__tests__/p4-sync.test.ts b/apps/api/src/lib/__tests__/p4-sync.test.ts index 7aeb0c87..4bf6b8e0 100644 --- a/apps/api/src/lib/__tests__/p4-sync.test.ts +++ b/apps/api/src/lib/__tests__/p4-sync.test.ts @@ -455,6 +455,32 @@ describe('executeP4Sync', () => { expect(result.message).toContain('Permission denied') }) + it('redacts credentials echoed back in the sync failure output', async () => { + makeSpawnMock(0) // p4Login succeeds + mockExecFile.mockImplementation((...args: unknown[]) => { + const cb = args[args.length - 1] as (err: Error & { stderr?: string; code?: string }) => void + const err = new Error('sync failed') as Error & { stderr?: string; code?: string } + // p4d echoes the connection string back on failure. + err.stderr = 'Connect to server failed; P4PASSWD=s3cr3t-token check $P4PORT' + err.code = '1' + cb(err) + }) + + const config: P4Config = { + ...p4ConfigDefaults, + p4port: 'ssl:h:1666', + p4user: 'u', + p4passwd: 's3cr3t-token', + p4client: 'c', + } + const result = await executeP4Sync(config, '/repo') + expect(result.ok).toBe(false) + // The message is persisted to lastSyncError and shipped to an outbound + // webhook, so it must never carry the password. + expect(result.message).not.toContain('s3cr3t-token') + expect(result.message).toContain('P4PASSWD=***') + }) + it('reports timeout on ETIMEDOUT', async () => { makeSpawnMock(0) // p4Login succeeds mockExecFile.mockImplementation((...args: unknown[]) => { diff --git a/apps/api/src/lib/agent-import.ts b/apps/api/src/lib/agent-import.ts index 0a119bad..e0a81def 100644 --- a/apps/api/src/lib/agent-import.ts +++ b/apps/api/src/lib/agent-import.ts @@ -1,4 +1,4 @@ -import { writeFileSync } from 'node:fs' +import { existsSync, rmSync, writeFileSync } from 'node:fs' import { dirname, join, resolve, sep } from 'node:path' import { chatAppConfigSchema, @@ -1020,6 +1020,34 @@ export async function importAgentFromZip( userId, }) + // Skill files are written INSIDE the transaction, as its last step. Writing + // them after the commit meant an ENOSPC/EDQUOT here left committed skills + // rows whose storagePath was empty or half-written — the operator saw a 500 + // and reasonably concluded nothing was imported, while every later run of + // the Agent silently mounted an incomplete skill package. Throwing here now + // rolls the rows back; the directories are cleaned up so a retry starts + // from a clean slate rather than adding an "(Imported)" duplicate set. + const createdDirs: string[] = [] + try { + for (const dir of pendingDirs) { + if (!existsSync(dir)) createdDirs.push(dir) + ensureDir(dir) + } + for (const { path, data } of pendingFileWrites) { + writeFileSync(path, data) + } + } catch (err) { + for (const dir of createdDirs) { + try { + rmSync(dir, { recursive: true, force: true }) + } catch { + // Best-effort: the rethrown error is what the operator acts on, and + // the transaction rollback is what keeps the database consistent. + } + } + throw err + } + return { agent: { id: agentId, name: agentName }, mcpServers: importedMcps, @@ -1028,14 +1056,6 @@ export async function importAgentFromZip( } }) - // Transaction succeeded — now write skill files to disk - for (const dir of pendingDirs) { - ensureDir(dir) - } - for (const { path, data } of pendingFileWrites) { - writeFileSync(path, data) - } - return result } diff --git a/apps/api/src/lib/feishu-service.ts b/apps/api/src/lib/feishu-service.ts index a8a92aa1..c98aec15 100644 --- a/apps/api/src/lib/feishu-service.ts +++ b/apps/api/src/lib/feishu-service.ts @@ -2143,7 +2143,11 @@ class FeishuConnectionManager { const executeJob = async () => { let imageTempDir: string | undefined let fileTempDir: string | undefined - let rootTempDir: string | undefined + // Tracked separately for the same reason imageTempDir/fileTempDir are: a + // root message can carry both images and a file, and a single variable + // would leak whichever download root it was overwritten with. + let rootImageTempDir: string | undefined + let rootFileTempDir: string | undefined let lifecycleParams: | { taskId: string @@ -2222,7 +2226,7 @@ class FeishuConnectionManager { 'Feishu: extracting images from topic root message', ) const rootDownloadRootDir = buildFeishuImageDownloadRootDir(agentId, runId) - rootTempDir = rootDownloadRootDir + rootImageTempDir = rootDownloadRootDir const rootDownloadDir = buildFeishuMessageResourceDownloadDir(rootDownloadRootDir) const downloaded = await downloadFeishuImages( freshClient, @@ -2240,7 +2244,7 @@ class FeishuConnectionManager { rootMeta.fileKey, ) const rootDownloadRootDir = buildFeishuFileDownloadRootDir(agentId, runId) - rootTempDir = rootDownloadRootDir + rootFileTempDir = rootDownloadRootDir const rootDownloadDir = buildFeishuMessageResourceDownloadDir(rootDownloadRootDir) const rootDl = await downloadFeishuFile( freshClient, @@ -2815,9 +2819,11 @@ class FeishuConnectionManager { ) } } finally { - if (imageTempDir) cleanupFeishuMessageResourceDownloadRoot(imageTempDir) - if (fileTempDir) cleanupFeishuMessageResourceDownloadRoot(fileTempDir) - if (rootTempDir) cleanupFeishuMessageResourceDownloadRoot(rootTempDir) + // Awaited: the run must not be reported terminal while its download + // roots are still being torn down. + for (const dir of [imageTempDir, fileTempDir, rootImageTempDir, rootFileTempDir]) { + if (dir) await cleanupFeishuMessageResourceDownloadRoot(dir) + } // Event has reached a terminal state (success or failure) — the DB // tombstone is no longer needed for restart recovery. await removePending() diff --git a/apps/api/src/lib/memory-topics.ts b/apps/api/src/lib/memory-topics.ts index 097166cc..48b59f9f 100644 --- a/apps/api/src/lib/memory-topics.ts +++ b/apps/api/src/lib/memory-topics.ts @@ -1059,8 +1059,30 @@ export function mergeMemoryTopics( ].slice(0, MEMORY_TOPIC_KEYWORD_LIMIT), updatedAt: new Date().toISOString(), } - writeMemoryFile(agentId, target.path, renderMemoryTopicFile(metadata, mergedBody)) - for (const source of sources) archiveMemoryTopic(agentId, source.topicId) + // The target is rewritten destructively and the sources are archived one by + // one, so a failure partway through would otherwise leave the target holding + // a source's facts while that source is still active — duplicated content + // that also makes the retry fail permanently (the archived sources no longer + // read back). Restore everything on the way out, the way splitMemoryTopic does. + const targetBefore = readMemoryFile(agentId, target.path) + const archived: string[] = [] + try { + writeMemoryFile(agentId, target.path, renderMemoryTopicFile(metadata, mergedBody)) + for (const source of sources) { + archiveMemoryTopic(agentId, source.topicId) + archived.push(source.topicId) + } + } catch (err) { + try { + for (const topicId of archived) reactivateMemoryTopic(agentId, topicId) + writeMemoryFile(agentId, target.path, targetBefore) + rebuildMemoryMain(agentId) + } catch { + // Preserve the original error; repair tooling can recover from the + // archived copies, which are written before their source is removed. + } + throw err + } rebuildMemoryMain(agentId) return readMemoryTopic(agentId, targetTopicId) } diff --git a/apps/api/src/lib/p4-sync.ts b/apps/api/src/lib/p4-sync.ts index cc1e6bc2..d264a077 100644 --- a/apps/api/src/lib/p4-sync.ts +++ b/apps/api/src/lib/p4-sync.ts @@ -298,7 +298,10 @@ export async function executeP4Sync( else if (stdout) parts.push(stdout) else if (parts.length === 0) parts.push(error instanceof Error ? error.message : String(error)) - return { ok: false, message: `P4 sync failed: ${parts.join(' — ')}` } + // Same redaction the connection check applies: p4d can echo the connection + // string (including P4PASSWD) back on failure, and this message is persisted + // to scmSources.lastSyncError and shipped to the sync-error webhook. + return { ok: false, message: `P4 sync failed: ${sanitizeCredentials(parts.join(' — '))}` } } } diff --git a/apps/api/src/routes/agents.ts b/apps/api/src/routes/agents.ts index a0569e11..5abd1371 100644 --- a/apps/api/src/routes/agents.ts +++ b/apps/api/src/routes/agents.ts @@ -2335,6 +2335,11 @@ app.post('/:id/regenerate-api-key', async (c) => { .update(agents) .set({ endpointApiKey: newApiKey, updatedAt: new Date() }) .where(eq(agents.id, id)) + + // Rotating a live credential breaks every integration holding the old key; + // the trail is exactly what answers "who rotated this, and when". + logAudit(c, { action: 'agent.regenerate_api_key', resource: 'agent', resourceId: id }) + return c.json({ data: { endpointApiKey: newApiKey } }) }) @@ -2350,6 +2355,9 @@ app.post('/:id/regenerate-a2a-api-key', async (c) => { .update(agents) .set({ a2aEndpointApiKey: newApiKey, updatedAt: new Date() }) .where(eq(agents.id, id)) + + logAudit(c, { action: 'agent.regenerate_a2a_api_key', resource: 'agent', resourceId: id }) + return c.json({ data: { a2aEndpointApiKey: newApiKey } }) }) diff --git a/apps/api/src/routes/runs.ts b/apps/api/src/routes/runs.ts index 0e1f881a..de6497c3 100644 --- a/apps/api/src/routes/runs.ts +++ b/apps/api/src/routes/runs.ts @@ -1035,6 +1035,15 @@ app.post('/:id/rerun', async (c) => { .returning() )[0] + const slotResult = await tryAcquireSlot(taskQueueDb, agentId, newRunId, agent.maxConcurrency ?? 1) + if (slotResult === 'queue_full') { + await db.delete(runs).where(eq(runs.id, newRunId)) + return c.json({ error: 'Queue is full' }, 429) + } + + // Audit after the slot was acquired, so a request rejected for a full queue + // (whose run row is deleted again) leaves no entry claiming a rerun that + // never happened — the same ordering POST /runs/:id/execute uses. logAudit(c, { action: 'run.rerun', resource: 'run', @@ -1042,11 +1051,6 @@ app.post('/:id/rerun', async (c) => { details: { originalRunId: id }, }) - const slotResult = await tryAcquireSlot(taskQueueDb, agentId, newRunId, agent.maxConcurrency ?? 1) - if (slotResult === 'queue_full') { - await db.delete(runs).where(eq(runs.id, newRunId)) - return c.json({ error: 'Queue is full' }, 429) - } if (slotResult === 'queued') { if (rerunContext) registerPendingContext(newRunId, rerunContext) return c.json({ data: { ...newRun, status: 'queued' } }, 202) diff --git a/apps/api/src/routes/scm-sources.ts b/apps/api/src/routes/scm-sources.ts index 2d7154ef..cc3eda43 100644 --- a/apps/api/src/routes/scm-sources.ts +++ b/apps/api/src/routes/scm-sources.ts @@ -940,6 +940,17 @@ app.post('/:id/sync', async (c) => { return c.json({ error: 'Sync already in progress' }, 409) } + // Audited after the status CAS and the checkout lock both won, so a request + // rejected with 409 leaves no entry claiming a sync that never started. The + // background job dials an external host with the row's stored credential and + // carries no actor of its own, so this is the only record of who triggered it. + logAudit(c, { + action: 'scm_source.sync', + resource: 'scm_source', + resourceId: id, + details: { type: source.type }, + }) + syncScmSource(id, { statusAlreadyAcquired: true, checkoutAlreadyAcquired: true }).catch((err) => { logger.error({ sourceId: id, error: err }, 'Background sync failed') }) @@ -959,15 +970,35 @@ app.post('/:id/check', async (c) => { return c.json({ error: 'SCM source not found' }, 404) } + // Same reasoning as POST /probe: this makes an outbound connection with the + // row's stored credential and writes no other record, so the audit entry is + // the only trace it happened (Iron Rule 5). Endpoint recorded redacted. + const auditCheck = (ok: boolean) => + logAudit(c, { + action: 'scm_source.check', + resource: 'scm_source', + resourceId: id, + details: { + type: source.type, + endpoint: probeEndpointForAudit({ + ...(source.config as unknown as P4Config | GitConfig), + type: source.type, + } as Parameters[0]), + ok, + }, + }) + if (source.type === 'p4') { const config = source.config as unknown as P4Config const result = await checkP4Connection(config, source.localPath) + auditCheck(result.ok) return c.json({ data: result }) } if (source.type === 'git') { const config = source.config as unknown as GitConfig const result = await checkGitConnection(config) + auditCheck(result.ok) return c.json({ data: result }) } diff --git a/apps/web/src/locales/en.json b/apps/web/src/locales/en.json index 10272f54..dbb678d7 100644 --- a/apps/web/src/locales/en.json +++ b/apps/web/src/locales/en.json @@ -2207,6 +2207,8 @@ "agent.pin": "Pin Agent", "agent.publish": "Publish Agent", "agent.publish_channel": "Save publish channel config", + "agent.regenerate_a2a_api_key": "Regenerate A2A API Key", + "agent.regenerate_api_key": "Regenerate API Key", "agent.restricted_mcp_blocked": "Restricted MCP blocked", "agent.resume": "Resume Agent", "agent.share": "Share Agent", @@ -2263,7 +2265,9 @@ "scm_source.request_deletion": "Request SCM source deletion", "scm_source.reclaim_storage": "Reclaim SCM source managed storage", "scm_source.probe": "Probe SCM source connectivity", + "scm_source.check": "Check SCM source connectivity", "scm_source.setup-script.run": "Run SCM setup script", + "scm_source.sync": "Trigger SCM source sync", "scm_source.update": "Update SCM source", "scm_source.workspace.delete": "Delete SCM workspace", "settings.auth.updated": "Update login settings", diff --git a/apps/web/src/locales/zh.json b/apps/web/src/locales/zh.json index 442c7025..df292f13 100644 --- a/apps/web/src/locales/zh.json +++ b/apps/web/src/locales/zh.json @@ -2234,6 +2234,8 @@ "agent.pin": "置顶 Agent", "agent.publish": "发布 Agent", "agent.publish_channel": "保存发布渠道配置", + "agent.regenerate_a2a_api_key": "重新生成 A2A API Key", + "agent.regenerate_api_key": "重新生成 API Key", "agent.restricted_mcp_blocked": "受限 MCP 已拦截", "agent.resume": "恢复 Agent", "agent.share": "分享 Agent", @@ -2290,7 +2292,9 @@ "scm_source.request_deletion": "申请删除代码源", "scm_source.reclaim_storage": "回收代码源托管存储", "scm_source.probe": "探测代码源连通性", + "scm_source.check": "检查代码源连通性", "scm_source.setup-script.run": "执行代码源初始化脚本", + "scm_source.sync": "触发代码源同步", "scm_source.update": "更新代码源", "scm_source.workspace.delete": "删除代码源工作区", "settings.auth.updated": "更新登录设置",