feat: harden autonomy lifecycle, OOM bounds, and provider-boundary finalization

This PR consolidates a coordinated batch of fixes around autonomy run/flow lifecycle, scheduled task deduplication, provider-boundary state finalization, and matching memory-bound treatments for adjacent long-running subsystems (REPL fullscreen scrollback, skill-search/skill-learning runtime activation). All changes were developed and reviewed together because they touched the same lifecycle invariants and were uncovered by the same long-running session reproductions.

## Lifecycle correctness

- Queued autonomy prompts are not injected unless the persisted run was successfully claimed; queued run claiming is now terminal-safe so a once-consumed/cancelled/failed run can not slip back into `queued`.
- Autonomy run/flow finalization happens on completion, provider error, generator close, and cancellation — not just the happy path. New `src/__tests__/queryAutonomyProviderBoundary.test.ts` covers these provider-boundary transitions.
- `requestManagedAutonomyFlowCancel` and `resumeManagedAutonomyFlowPrompt` carry `rootDir` and `currentDir` explicitly across detached async boundaries (proactive-tick, cron, daemon restart) instead of inferring from process state.
- Active runs/flows are protected from janitor pruning so a running step can not be garbage-collected mid-flight (`src/utils/autonomyAuthority.ts`).
- Heartbeat parser now ignores fenced code blocks; the two-phase commit window for autonomy state transitions is documented in `docs/internals/autonomy-jira.md`.

## Ownership and dedup

- `src/utils/autonomyRuns.ts`: ownership stamping (run id + rootDir carried end-to-end), source-based dedup against active runs.
- `src/hooks/useScheduledTasks.ts`: scheduled ticks deduplicate against runs already active on the same source label.
- `src/utils/processUserInput/processSlashCommand.tsx`: forked slash commands now thread the autonomy `runId` so completion finalizers can find the originating run for deferred completion.
- New `src/utils/autonomyQueueLifecycle.ts` and tests collect the queue-side lifecycle invariants in one place.

## Memory bounds (related, same review pass)

- `src/screens/REPL.tsx`: caps fullscreen scrollback after the compact boundary and updates trailing progress rows in place. Long-running fullscreen sessions could otherwise retain thousands of post-compaction messages and duplicate progress rows, keeping Ink trees alive long after their useful context had moved on.
- `src/services/skillSearch/*` and `src/services/skillLearning/*`: runtime activation is strictly opt-in via existing env toggles; session caches are capped so long-running processes can not grow them forever. Build presence is preserved so operators can still discover and opt into the slash commands.

## CI / test contract

- `tests/integration/dependency-overrides.test.ts`: smoke test no longer drives Mermaid's browser renderer; it validates the package-resolution contract directly so CI does not regress on unrelated browser timing.
- New `tests/integration/autonomy-lifecycle-user-flow.test.ts`: end-to-end CLI subprocess flow exercising `status --deep`, `flows`, `flow <id>`, `flow resume`, `flow cancel` against persisted state.
- `src/entrypoints/cli.tsx`: `claude autonomy …` routes through an entrypoint fast path that reuses the slash-command formatter without booting the full interactive CLI. Stdout is flushed before forced exit so coverage subprocesses do not terminate with empty stdout.
- `packages/builtin-tools/src/tools/RemoteTriggerTool/__tests__/RemoteTriggerTool.test.ts`: stabilized to prevent audit flake under coverage.

## Tests added

- `src/__tests__/queryAutonomyProviderBoundary.test.ts`
- `src/hooks/__tests__/useScheduledTasks.test.ts`
- `src/utils/__tests__/autonomyAuthority.test.ts`
- `src/utils/__tests__/autonomyFlows.test.ts` (extended)
- `src/utils/__tests__/autonomyPersistence.test.ts` (extended)
- `src/utils/__tests__/autonomyQueueLifecycle.test.ts`
- `src/utils/__tests__/autonomyRuns.test.ts` (extended)
- `src/utils/processUserInput/__tests__/processSlashCommand.test.ts`
- `tests/integration/autonomy-lifecycle-user-flow.test.ts`

## Docs

- `docs/agent/sur-loop-scheduled-oom.md`: System Understanding Report covering the scheduled/loop OOM problem, the call graphs investigated, and the lifecycle invariants this PR establishes.
- `docs/agent/sur-skill-overflow-bugs.md`: SUR for the related skill-overflow context.
- `docs/internals/autonomy-jira.md`: documents the two-phase commit window and ownership stamping invariants.
- `docs/memory-leak-audit.md`: audit notes covering the REPL/scrollback and skill-search bounds.

## Invariants this PR establishes

1. Queued autonomy prompts are not injected unless the persisted run was successfully claimed.
2. Terminal run/flow states are terminal — completion, failure, and cancellation all finalize state regardless of which provider/error path triggered them.
3. Autonomy run/flow `rootDir` is carried explicitly across detached async boundaries instead of inferred from a shared singleton.
4. State-only CLI subcommands (`autonomy status|runs|flows|flow …`) bypass full interactive bootstrap so they do not hold unrelated handles open.
5. REPL fullscreen scrollback and skill-search/skill-learning session caches are explicitly bounded.

## Validation

```bash
bun run typecheck
CI=true GITHUB_ACTIONS=true bun test            # 3996 pass / 0 fail across 305 files
bun test src/__tests__/queryAutonomyProviderBoundary.test.ts \
         src/hooks/__tests__/useScheduledTasks.test.ts \
         src/utils/__tests__/autonomy{Runs,Flows,Authority,QueueLifecycle,Persistence}.test.ts \
         src/utils/processUserInput/__tests__/processSlashCommand.test.ts \
         tests/integration/autonomy-lifecycle-user-flow.test.ts
```

## Origin

This PR is the consolidated, upstream-targeted version of two fork-side review PRs (fix/loop-scheduled-autonomy-oom and fix/autonomy-lifecycle). The fork-side review history is preserved at https://github.com/amDosion/claude-code-bast/pull/7 . The fork's own internal `chore: keep fork current with upstream` sync commits and the `docs: update contributors` automation are intentionally not included in this PR.

The autonomy CLI handler `rootDir` threading that the fork added (78f64d8a, 98d04ddb) is intentionally omitted here because upstream `a2cfaf91` (fix: 修复 RemoteTriggerTool 和 autonomy 测试的全量运行失败) already performed the equivalent change with an additional `currentDir` option. Keeping the upstream version avoids regressing that improvement.

src/utils/autonomyRuns.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from 'crypto'
 import { mkdir, writeFile } from 'fs/promises'
 import { dirname, join, resolve } from 'path'
-import { getProjectRoot } from '../bootstrap/state.js'
+import { getProjectRoot, getSessionId } from '../bootstrap/state.js'
 import type { MessageOrigin } from '../types/message.js'
 import type { QueuedCommand } from '../types/textInputTypes.js'
 import {
@@ -29,9 +29,22 @@ import {
 } from './autonomyFlows.js'
 import { withAutonomyPersistenceLock } from './autonomyPersistence.js'
 import { getFsImplementation } from './fsOperations.js'
+import { isProcessRunning } from './genericProcessUtils.js'
+import { logError } from './log.js'
 
 const AUTONOMY_RUNS_MAX = 200
 const AUTONOMY_RUNS_RELATIVE_PATH = join(AUTONOMY_DIR, 'runs.json')
+// Sentinel string surfaced to operators via runs.json error fields and
+// referenced literally by the HEARTBEAT.md `stale-recovery-health` task.
+// A unit test asserts the HEARTBEAT.md file contains this exact prefix —
+// changing the value will fail the test, forcing the heartbeat prompt
+// to be updated in the same change.
+export const STALE_ACTIVE_RUN_ERROR_PREFIX =
+  'Recovered stale active autonomy run'
+
+// Guards the legacy-block warning so it fires once per (process, runId) instead
+// of every dedup tick while a no-owner record sits there.
+const warnedLegacyBlockRunIds = new Set<string>()
 
 export type AutonomyRunStatus =
   | 'queued'
@@ -59,6 +72,8 @@ export type AutonomyRunRecord = {
   flowStepName?: string
   promptPreview: string
   createdAt: number
+  ownerProcessId?: number
+  ownerSessionId?: string
   startedAt?: number
   endedAt?: number
   error?: string
@@ -77,6 +92,19 @@ type AutonomyRunFlowRef = {
   stepName: string
 }
 
+type CreateAutonomyRunParams = {
+  trigger: AutonomyTriggerKind
+  prompt: string
+  rootDir?: string
+  currentDir?: string
+  sourceId?: string
+  sourceLabel?: string
+  runtime?: AutonomyRunRuntime
+  ownerKey?: string
+  flow?: AutonomyRunFlowRef
+  nowMs?: number
+}
+
 function truncatePromptPreview(prompt: string): string {
   const singleLine = prompt.replace(/\s+/g, ' ').trim()
   return singleLine.length <= 240
@@ -95,6 +123,29 @@ function cloneRunRecord(run: AutonomyRunRecord): AutonomyRunRecord {
   return { ...run }
 }
 
+function isAutonomyRunActive(run: AutonomyRunRecord): boolean {
+  return run.status === 'queued' || run.status === 'running'
+}
+
+function selectPersistedAutonomyRuns(
+  runs: AutonomyRunRecord[],
+): AutonomyRunRecord[] {
+  const retained = runs
+    .slice()
+    .map(cloneRunRecord)
+    .sort((left, right) => {
+      const leftActive = isAutonomyRunActive(left)
+      const rightActive = isAutonomyRunActive(right)
+      if (leftActive !== rightActive) {
+        return leftActive ? -1 : 1
+      }
+      return right.createdAt - left.createdAt
+    })
+    .slice(0, AUTONOMY_RUNS_MAX)
+
+  return retained.sort((left, right) => right.createdAt - left.createdAt)
+}
+
 function normalizePersistedRunRecord(
   run: PersistedAutonomyRunRecord,
 ): AutonomyRunRecord {
@@ -157,11 +208,7 @@ async function writeAutonomyRuns(
     path,
     `${JSON.stringify(
       {
-        runs: runs
-          .slice()
-          .map(cloneRunRecord)
-          .sort((left, right) => right.createdAt - left.createdAt)
-          .slice(0, AUTONOMY_RUNS_MAX),
+        runs: selectPersistedAutonomyRuns(runs),
       } satisfies AutonomyRunsFile,
       null,
       2,
@@ -172,7 +219,7 @@ async function writeAutonomyRuns(
 
 async function updateAutonomyRun(
   runId: string,
-  updater: (current: AutonomyRunRecord) => AutonomyRunRecord,
+  updater: (current: AutonomyRunRecord) => AutonomyRunRecord | null,
   rootDir: string = getProjectRoot(),
 ): Promise<AutonomyRunRecord | null> {
   return withAutonomyPersistenceLock(rootDir, async () => {
@@ -181,7 +228,11 @@ async function updateAutonomyRun(
     if (index === -1) {
       return null
     }
-    const updated = cloneRunRecord(updater(cloneRunRecord(runs[index]!)))
+    const next = updater(cloneRunRecord(runs[index]!))
+    if (!next) {
+      return null
+    }
+    const updated = cloneRunRecord(next)
     runs[index] = updated
     await writeAutonomyRuns(runs, rootDir)
     return updated
@@ -196,21 +247,112 @@ export async function getAutonomyRunById(
   return runs.find(run => run.runId === runId) ?? null
 }
 
-export async function createAutonomyRun(params: {
+function isActiveAutonomyRunStatus(status: AutonomyRunStatus): boolean {
+  return status === 'queued' || status === 'running'
+}
+
+function isValidOwnerProcessId(pid: number | undefined): pid is number {
+  // Reject non-numeric, negative, zero (Linux: send-to-process-group), and
+  // non-integer values. A forged record with pid=0 or pid<0 used to be
+  // treated as live and could permanently block dedup; treating them as
+  // stale closes that availability hole.
+  return (
+    typeof pid === 'number' &&
+    Number.isInteger(pid) &&
+    pid > 0 &&
+    pid < 4_194_304
+  )
+}
+
+function isStaleActiveAutonomyRun(run: AutonomyRunRecord): boolean {
+  if (!isActiveAutonomyRunStatus(run.status)) {
+    return false
+  }
+  if (run.ownerProcessId === undefined) {
+    return false
+  }
+  if (!isValidOwnerProcessId(run.ownerProcessId)) {
+    return true
+  }
+  return !isProcessRunning(run.ownerProcessId)
+}
+
+function staleActiveRunError(run: AutonomyRunRecord): string {
+  return `${STALE_ACTIVE_RUN_ERROR_PREFIX}: owner process ${run.ownerProcessId} is no longer running.`
+}
+
+function failAutonomyRunRecord(
+  run: AutonomyRunRecord,
+  error: string,
+  nowMs: number,
+): AutonomyRunRecord {
+  return {
+    ...run,
+    status: 'failed',
+    endedAt: nowMs,
+    error,
+  }
+}
+
+function recoverStaleActiveAutonomyRun(
+  run: AutonomyRunRecord,
+  nowMs: number,
+): AutonomyRunRecord {
+  return failAutonomyRunRecord(run, staleActiveRunError(run), nowMs)
+}
+
+async function syncFailedManagedFlowForRun(
+  run: AutonomyRunRecord,
+  rootDir: string,
+): Promise<void> {
+  if (run.parentFlowId && run.parentFlowSyncMode === 'managed') {
+    await markManagedAutonomyFlowStepFailed({
+      flowId: run.parentFlowId,
+      runId: run.runId,
+      error: run.error ?? 'Autonomy run failed.',
+      rootDir,
+      nowMs: run.endedAt,
+    })
+  }
+}
+
+function matchesActiveAutonomyRunSource(
+  run: AutonomyRunRecord,
+  params: {
+    trigger: AutonomyTriggerKind
+    sourceId: string
+    ownerKey?: string
+  },
+): boolean {
+  return (
+    run.trigger === params.trigger &&
+    run.sourceId === params.sourceId &&
+    (params.ownerKey === undefined || run.ownerKey === params.ownerKey) &&
+    isActiveAutonomyRunStatus(run.status)
+  )
+}
+
+export async function hasActiveAutonomyRunForSource(params: {
   trigger: AutonomyTriggerKind
-  prompt: string
+  sourceId: string
   rootDir?: string
-  currentDir?: string
-  sourceId?: string
-  sourceLabel?: string
-  runtime?: AutonomyRunRuntime
   ownerKey?: string
-  flow?: AutonomyRunFlowRef
-  nowMs?: number
-}): Promise<AutonomyRunRecord> {
-  const rootDir = resolve(params.rootDir ?? getProjectRoot())
-  const currentDir = resolve(params.currentDir ?? rootDir)
-  const record: AutonomyRunRecord = {
+}): Promise<boolean> {
+  const runs = await listAutonomyRuns(params.rootDir)
+  return runs.some(
+    run =>
+      matchesActiveAutonomyRunSource(run, params) &&
+      !isStaleActiveAutonomyRun(run),
+  )
+}
+
+function buildAutonomyRunRecord(
+  params: CreateAutonomyRunParams,
+  rootDir: string,
+  currentDir: string,
+): AutonomyRunRecord {
+  const createdAt = params.nowMs ?? Date.now()
+  return {
     runId: randomUUID(),
     runtime: params.runtime ?? (params.flow ? 'flow_step' : 'automatic'),
     trigger: params.trigger,
@@ -231,13 +373,80 @@ export async function createAutonomyRun(params: {
         }
       : {}),
     promptPreview: truncatePromptPreview(params.prompt),
-    createdAt: params.nowMs ?? Date.now(),
+    createdAt,
+    ownerProcessId: process.pid,
+    ownerSessionId: getSessionId(),
   }
+}
+
+async function persistAutonomyRunRecord(
+  record: AutonomyRunRecord,
+  rootDir: string,
+  skipWhenActiveSource: boolean,
+): Promise<{
+  created: boolean
+  recoveredStaleRuns: AutonomyRunRecord[]
+}> {
+  let created = false
+  const recoveredStaleRuns: AutonomyRunRecord[] = []
   await withAutonomyPersistenceLock(rootDir, async () => {
     const runs = await listAutonomyRuns(rootDir)
+    const sourceId = record.sourceId
+    if (skipWhenActiveSource && sourceId) {
+      let hasBlockingActiveRun = false
+      let staleRecoveriesApplied = false
+      for (let i = 0; i < runs.length; i++) {
+        const run = runs[i]!
+        if (
+          !matchesActiveAutonomyRunSource(run, {
+            trigger: record.trigger,
+            sourceId,
+            ownerKey: record.ownerKey,
+          })
+        ) {
+          continue
+        }
+        if (isStaleActiveAutonomyRun(run)) {
+          const recovered = recoverStaleActiveAutonomyRun(
+            run,
+            record.createdAt,
+          )
+          runs[i] = recovered
+          recoveredStaleRuns.push(recovered)
+          staleRecoveriesApplied = true
+          continue
+        }
+        if (
+          run.ownerProcessId === undefined &&
+          !warnedLegacyBlockRunIds.has(run.runId)
+        ) {
+          warnedLegacyBlockRunIds.add(run.runId)
+          logError(
+            new Error(
+              `[autonomyRuns] blocked by legacy un-owned active run ${run.runId} (createdAt=${run.createdAt}); cancel manually if this is a stale upgrade artifact`,
+            ),
+          )
+        }
+        hasBlockingActiveRun = true
+      }
+      if (hasBlockingActiveRun) {
+        if (staleRecoveriesApplied) {
+          await writeAutonomyRuns(runs, rootDir)
+        }
+        return
+      }
+    }
     runs.unshift(record)
     await writeAutonomyRuns(runs, rootDir)
+    created = true
   })
+  return { created, recoveredStaleRuns }
+}
+
+async function queueManagedFlowStepRunForRecord(
+  record: AutonomyRunRecord,
+  rootDir: string,
+): Promise<void> {
   if (
     record.parentFlowId &&
     record.flowStepId &&
@@ -258,9 +467,47 @@ export async function createAutonomyRun(params: {
       nowMs: record.createdAt,
     })
   }
+}
+
+async function createAutonomyRunCore(
+  params: CreateAutonomyRunParams,
+  skipIfActiveSource: boolean,
+): Promise<AutonomyRunRecord | null> {
+  const rootDir = resolve(params.rootDir ?? getProjectRoot())
+  const currentDir = resolve(params.currentDir ?? rootDir)
+  const record = buildAutonomyRunRecord(params, rootDir, currentDir)
+
+  const { created, recoveredStaleRuns } = await persistAutonomyRunRecord(
+    record,
+    rootDir,
+    skipIfActiveSource,
+  )
+  for (const recovered of recoveredStaleRuns) {
+    await syncFailedManagedFlowForRun(recovered, rootDir)
+  }
+  if (!created) {
+    return null
+  }
+  await queueManagedFlowStepRunForRecord(record, rootDir)
   return record
 }
 
+export async function createAutonomyRun(
+  params: CreateAutonomyRunParams,
+): Promise<AutonomyRunRecord> {
+  const record = await createAutonomyRunCore(params, false)
+  if (!record) {
+    throw new Error('Autonomy run was unexpectedly skipped.')
+  }
+  return record
+}
+
+export async function createAutonomyRunIfNoActiveSource(
+  params: CreateAutonomyRunParams & { sourceId: string },
+): Promise<AutonomyRunRecord | null> {
+  return createAutonomyRunCore(params, true)
+}
+
 function buildManagedFlowStepPrompt(
   flow: AutonomyFlowRecord,
   stepIndex: number,
@@ -336,6 +583,7 @@ async function createOrRecoverManagedFlowStepCommand(params: {
         workload: params.workload,
         autonomy: {
           runId: run.runId,
+          rootDir: run.rootDir,
           trigger: 'managed-flow-step',
           sourceId: run.sourceId,
           sourceLabel: run.sourceLabel,
@@ -426,11 +674,16 @@ export async function markAutonomyRunRunning(
 ): Promise<AutonomyRunRecord | null> {
   const updated = await updateAutonomyRun(
     runId,
-    current => ({
-      ...current,
-      status: 'running',
-      startedAt: nowMs ?? Date.now(),
-    }),
+    current =>
+      current.status === 'queued'
+        ? {
+            ...current,
+            status: 'running',
+            startedAt: nowMs ?? Date.now(),
+            ownerProcessId: process.pid,
+            ownerSessionId: getSessionId(),
+          }
+        : null,
     rootDir,
   )
   if (updated?.parentFlowId && updated.parentFlowSyncMode === 'managed') {
@@ -451,12 +704,15 @@ export async function markAutonomyRunCompleted(
 ): Promise<AutonomyRunRecord | null> {
   const updated = await updateAutonomyRun(
     runId,
-    current => ({
-      ...current,
-      status: 'completed',
-      endedAt: nowMs ?? Date.now(),
-      error: undefined,
-    }),
+    current =>
+      current.status === 'queued' || current.status === 'running'
+        ? {
+            ...current,
+            status: 'completed',
+            endedAt: nowMs ?? Date.now(),
+            error: undefined,
+          }
+        : null,
     rootDir,
   )
   if (updated?.parentFlowId && updated.parentFlowSyncMode === 'managed') {
@@ -476,24 +732,17 @@ export async function markAutonomyRunFailed(
   rootDir?: string,
   nowMs?: number,
 ): Promise<AutonomyRunRecord | null> {
+  const endedAt = nowMs ?? Date.now()
   const updated = await updateAutonomyRun(
     runId,
-    current => ({
-      ...current,
-      status: 'failed',
-      endedAt: nowMs ?? Date.now(),
-      error,
-    }),
+    current =>
+      isActiveAutonomyRunStatus(current.status)
+        ? failAutonomyRunRecord(current, error, endedAt)
+        : null,
     rootDir,
   )
-  if (updated?.parentFlowId && updated.parentFlowSyncMode === 'managed') {
-    await markManagedAutonomyFlowStepFailed({
-      flowId: updated.parentFlowId,
-      runId: updated.runId,
-      error,
-      rootDir,
-      nowMs: updated.endedAt,
-    })
+  if (updated) {
+    await syncFailedManagedFlowForRun(updated, rootDir ?? updated.rootDir)
   }
   return updated
 }
@@ -505,12 +754,15 @@ export async function markAutonomyRunCancelled(
 ): Promise<AutonomyRunRecord | null> {
   const updated = await updateAutonomyRun(
     runId,
-    current => ({
-      ...current,
-      status: 'cancelled',
-      endedAt: nowMs ?? Date.now(),
-      error: undefined,
-    }),
+    current =>
+      current.status === 'queued' || current.status === 'running'
+        ? {
+            ...current,
+            status: 'cancelled',
+            endedAt: nowMs ?? Date.now(),
+            error: undefined,
+          }
+        : null,
     rootDir,
   )
   if (updated?.parentFlowId && updated.parentFlowSyncMode === 'managed') {
@@ -612,6 +864,7 @@ export async function createAutonomyQueuedPrompt(params: {
   currentDir?: string
   sourceId?: string
   sourceLabel?: string
+  ownerKey?: string
   workload?: string
   priority?: 'now' | 'next' | 'later'
   shouldCreate?: () => boolean
@@ -634,39 +887,130 @@ export async function createAutonomyQueuedPrompt(params: {
     currentDir,
     sourceId: params.sourceId,
     sourceLabel: params.sourceLabel,
+    ownerKey: params.ownerKey,
     workload: params.workload,
     priority: params.priority,
     flow: params.flow,
   })
 }
 
+export async function createAutonomyQueuedPromptIfNoActiveSource(params: {
+  trigger: AutonomyTriggerKind
+  basePrompt: string
+  rootDir?: string
+  currentDir?: string
+  sourceId: string
+  sourceLabel?: string
+  ownerKey?: string
+  workload?: string
+  priority?: 'now' | 'next' | 'later'
+  shouldCreate?: () => boolean
+}): Promise<QueuedCommand | null> {
+  const rootDir = resolve(params.rootDir ?? getProjectRoot())
+  const currentDir = resolve(params.currentDir ?? getCwd())
+  // Cheap optimistic pre-check: skip the AGENTS.md / HEARTBEAT.md disk
+  // reads + prompt assembly when an active run for this source already
+  // blocks dedup. The lock-side check inside persistAutonomyRunRecord
+  // remains authoritative; this only fast-paths the common storm case.
+  if (
+    await hasActiveAutonomyRunForSource({
+      trigger: params.trigger,
+      sourceId: params.sourceId,
+      rootDir,
+      ownerKey: params.ownerKey,
+    })
+  ) {
+    return null
+  }
+  const prepared = await prepareAutonomyTurnPrompt({
+    basePrompt: params.basePrompt,
+    trigger: params.trigger,
+    rootDir,
+    currentDir,
+  })
+  if (params.shouldCreate && !params.shouldCreate()) {
+    return null
+  }
+  return commitAutonomyQueuedPromptIfNoActiveSource({
+    prepared,
+    rootDir,
+    currentDir,
+    sourceId: params.sourceId,
+    sourceLabel: params.sourceLabel,
+    ownerKey: params.ownerKey,
+    workload: params.workload,
+    priority: params.priority,
+  })
+}
+
 export async function commitAutonomyQueuedPrompt(params: {
   prepared: Awaited<ReturnType<typeof prepareAutonomyTurnPrompt>>
   rootDir?: string
   currentDir?: string
   sourceId?: string
   sourceLabel?: string
+  ownerKey?: string
   workload?: string
   priority?: 'now' | 'next' | 'later'
   flow?: AutonomyRunFlowRef
 }): Promise<QueuedCommand> {
+  const command = await commitAutonomyQueuedPromptInternal(params, false)
+  if (!command) {
+    throw new Error('Autonomy queued prompt was unexpectedly skipped.')
+  }
+  return command
+}
+
+async function commitAutonomyQueuedPromptIfNoActiveSource(params: {
+  prepared: Awaited<ReturnType<typeof prepareAutonomyTurnPrompt>>
+  rootDir?: string
+  currentDir?: string
+  sourceId: string
+  sourceLabel?: string
+  ownerKey?: string
+  workload?: string
+  priority?: 'now' | 'next' | 'later'
+}): Promise<QueuedCommand | null> {
+  return commitAutonomyQueuedPromptInternal(params, true)
+}
+
+async function commitAutonomyQueuedPromptInternal(
+  params: {
+    prepared: Awaited<ReturnType<typeof prepareAutonomyTurnPrompt>>
+    rootDir?: string
+    currentDir?: string
+    sourceId?: string
+    sourceLabel?: string
+    ownerKey?: string
+    workload?: string
+    priority?: 'now' | 'next' | 'later'
+    flow?: AutonomyRunFlowRef
+  },
+  skipWhenActiveSource: boolean,
+): Promise<QueuedCommand | null> {
   const rootDir = resolve(
     params.rootDir ?? params.prepared.rootDir ?? getProjectRoot(),
   )
   const currentDir = resolve(
     params.currentDir ?? params.prepared.currentDir ?? getCwd(),
   )
-  commitPreparedAutonomyTurn(params.prepared)
   const value = params.prepared.prompt
-  const run = await createAutonomyRun({
+  const runParams: CreateAutonomyRunParams = {
     trigger: params.prepared.trigger,
     prompt: value,
     rootDir,
     currentDir,
     sourceId: params.sourceId,
     sourceLabel: params.sourceLabel,
+    ownerKey: params.ownerKey,
     flow: params.flow,
-  })
+  }
+  const useDedup = skipWhenActiveSource && Boolean(params.sourceId)
+  const run = await createAutonomyRunCore(runParams, useDedup)
+  if (!run) {
+    return null
+  }
+  commitPreparedAutonomyTurn(params.prepared)
   const origin = {
     kind: 'autonomy',
     trigger: params.prepared.trigger,
@@ -683,6 +1027,7 @@ export async function commitAutonomyQueuedPrompt(params: {
     workload: params.workload,
     autonomy: {
       runId: run.runId,
+      rootDir: run.rootDir,
       trigger: params.prepared.trigger,
       sourceId: params.sourceId,
       sourceLabel: params.sourceLabel,