fix(api): redact P4 sync errors, restore merge on failure, close audit gaps

apps/api/src/lib/__tests__/memory-topics.test.ts

@@ -14,16 +14,17 @@ 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,
   estimateMemoryTokens,
   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

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[]) => {

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
 }
 

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()

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)
 }

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(' — '))}` }
   }
 }
 

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 } })
 })
 

apps/api/src/routes/runs.ts

@@ -1035,18 +1035,22 @@ 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',
     resourceId: newRunId,
     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)

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<typeof probeEndpointForAudit>[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 })
   }
 

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",