(CODUX-220) Reuse execution rows for provider telemetry

src/services/cli-workflow-service.ts

@@ -492,6 +492,7 @@ export class CliWorkflowService {
             category,
             errorMessage: blocker,
           }, `cli:workflow:blocked:agent:${args.sessionId}`);
+          this.finalizeExecutionInvocation(ctx.executionInvocationId, "failed", finishedAt, blocker);
           return;
         }
         this.appendExecutionEvent(args, "cli_git_no_changes", {
@@ -511,6 +512,7 @@ export class CliWorkflowService {
           provider: args.provider,
           outcome: "no_changes",
         }, "cli:workflow:completed:no-changes");
+        this.finalizeExecutionInvocation(ctx.executionInvocationId, "completed", finishedAt);
         return;
       }
 
@@ -552,6 +554,7 @@ export class CliWorkflowService {
         outcome: "pushed",
         prUrl: prUrl || null,
       }, `cli:workflow:completed:${prUrl || "none"}`);
+      this.finalizeExecutionInvocation(ctx.executionInvocationId, "completed", finishedAt);
 
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
@@ -713,6 +716,12 @@ export class CliWorkflowService {
           message,
         });
       }
+      this.finalizeExecutionInvocation(
+        ctx.executionInvocationId,
+        abortController.signal.aborted ? "cancelled" : "failed",
+        finishedAt,
+        message,
+      );
     } finally {
       try {
         const cleanupResult = preserveWorkspaceForShutdown
@@ -895,6 +904,26 @@ export class CliWorkflowService {
     }
   }
 
+  private finalizeExecutionInvocation(
+    invocationId: string | undefined,
+    status: "completed" | "failed" | "cancelled",
+    finishedAt: string,
+    errorMessage?: string,
+  ): void {
+    if (!invocationId || !this.deps.executionRepository) {
+      return;
+    }
+    const invocation = this.deps.executionRepository.getExecutionInvocation(invocationId);
+    if (!invocation || (invocation.status !== "running" && invocation.status !== "paused")) {
+      return;
+    }
+    this.deps.executionRepository.updateExecutionInvocation(invocationId, {
+      status,
+      finishedAt,
+      ...(status === "completed" ? { errorMessage: null } : { errorMessage: errorMessage ?? null }),
+    });
+  }
+
   private isSprintRunCancelled(sprintRunId?: string | null): boolean {
     if (!sprintRunId || !this.deps.executionRepository) {
       return false;

src/services/cli-workflow/pipeline/execute-provider-stage.ts

@@ -108,6 +108,7 @@ export async function executeProviderStage(ctx: PipelineContext, providerPrompt:
     continueSessionId,
     openCodeBaselineRawUsageJson,
     invocationId: ctx.executionInvocationId,
+    finalizeExecutionInvocation: ctx.executionInvocationId ? false : undefined,
     workflowSettings: ctx.workflowSettings,
     repoPath: ctx.repoPath,
     gitPolicy: {

src/services/provider-concurrency-service.ts

@@ -97,11 +97,18 @@ export class ProviderConcurrencyService {
     limit: number,
     input: CreateProviderInvocationUsageInput,
     signal?: AbortSignal,
-    maxWaitMs?: number
+    maxWaitMs?: number,
+    executionInvocationId?: string,
   ): Promise<ProviderInvocationUsageRecord> {
     if (limit <= 0) {
       await this.reconcileStaleProviderInvocations(provider, true);
-      return this.deps.executionRepository.createProviderInvocationUsage(input);
+      if (signal?.aborted) {
+        throw signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason || "AbortSignal triggered"));
+      }
+      this.assertExecutionInvocationCanClaim(executionInvocationId);
+      const invocation = this.deps.executionRepository.createProviderInvocationUsage(input);
+      this.linkExecutionInvocation(executionInvocationId, invocation.id);
+      return invocation;
     }
 
     const startMs = Date.now();
@@ -120,8 +127,13 @@ export class ProviderConcurrencyService {
       await this.reconcileStaleProviderInvocations(provider, isFirstCheck);
       isFirstCheck = false;
 
+      // Keep the status check and synchronous repository claim in the same event-loop turn.
+      // Dashboard cancellation can therefore stop a waiting execution before it manufactures
+      // provider usage or consumes capacity.
+      this.assertExecutionInvocationCanClaim(executionInvocationId);
       const invocation = this.deps.executionRepository.tryCreateProviderInvocationUsage(input, limit);
       if (invocation) {
+        this.linkExecutionInvocation(executionInvocationId, invocation.id);
         return invocation;
       }
 
@@ -139,6 +151,25 @@ export class ProviderConcurrencyService {
     }
   }
 
+  private assertExecutionInvocationCanClaim(executionInvocationId: string | undefined): void {
+    if (!executionInvocationId) {
+      return;
+    }
+    const invocation = this.deps.executionRepository.getExecutionInvocation(executionInvocationId);
+    if (invocation?.status === "cancelled") {
+      throw new Error(`Execution invocation ${executionInvocationId} is ${invocation.status}; provider slot will not be claimed.`);
+    }
+  }
+
+  private linkExecutionInvocation(executionInvocationId: string | undefined, providerInvocationId: string): void {
+    if (!executionInvocationId) {
+      return;
+    }
+    this.deps.executionRepository.updateExecutionInvocation(executionInvocationId, {
+      providerInvocationId,
+    });
+  }
+
   /**
    * Attempts to claim a concurrency slot for the given provider atomically without waiting.
    * Returns the claimed invocation record, or null if the global cap is currently reached.

src/services/provider-execution-service.ts

@@ -278,6 +278,9 @@ export class ProviderExecutionService {
   async executeProvider(args: ExecutionProviderRunArgs): Promise<ProviderRunResult> {
     let execInvocationId: string | null = args.invocationId || null;
     let lastPersistedMessagesSignature: string | null = null;
+    if (execInvocationId) {
+      this.assertExecutionInvocationCanRun(execInvocationId);
+    }
     const effectiveModel = resolveEffectiveModel(args);
     const scopedSettings = args.projectId.trim() && this.deps.getDashboardSettings
       ? this.deps.getDashboardSettings({
@@ -315,7 +318,10 @@ export class ProviderExecutionService {
     });
 
     const runProviderInner = async (p: string, retrySystemMessage?: string, continueSessionId?: string | null, openCodeBaselineRawUsageJson?: Record<string, unknown> | null): Promise<ProviderRunResult> => {
-      const startedAt = new Date().toISOString();
+      if (execInvocationId) {
+        this.assertExecutionInvocationCanRun(execInvocationId);
+      }
+      const executionStartedAt = new Date().toISOString();
 
       // Coalesce the per-line streaming activity firehose into batched transactions so concurrent
       // sprints don't saturate the single thread with one INSERT per output line. Only used when
@@ -347,19 +353,19 @@ export class ProviderExecutionService {
           type: args.type,
           provider: args.provider,
           model: effectiveModel,
-          startedAt,
+          startedAt: executionStartedAt,
           invocationSource: args.invocationSource,
         })?.id || null;
       }
 
-      if (execInvocationId && retrySystemMessage) {
+      if (execInvocationId && retrySystemMessage && this.isExecutionInvocationStillRunning(execInvocationId)) {
         this.deps.executionRepository?.appendExecutionInvocationMessage(execInvocationId, {
           role: "system",
           contentMarkdown: retrySystemMessage,
         });
       }
 
-      if (execInvocationId && args.trackPromptInInvocation !== false) {
+      if (execInvocationId && args.trackPromptInInvocation !== false && this.isExecutionInvocationStillRunning(execInvocationId)) {
         this.deps.executionRepository?.appendExecutionInvocationMessage(execInvocationId, {
           role: "user",
           contentMarkdown: p,
@@ -384,7 +390,6 @@ export class ProviderExecutionService {
         purpose: args.purpose,
         model: effectiveModel,
         executionMode: args.workflowSettings.executionMode,
-        startedAt,
         promptChars: p.length,
       };
 
@@ -395,20 +400,26 @@ export class ProviderExecutionService {
           usageInput,
           args.signal,
           args.concurrencyWaitTimeoutMs,
+          execInvocationId ?? undefined,
         );
       } else {
         // Fallback for cases where ProviderConcurrencyService is not provided, 
         // e.g. in some specialized service tests, though in production it should be present
         // when an execution repository is present.
+        if (execInvocationId) {
+          this.assertExecutionInvocationCanRun(execInvocationId);
+        }
         invocation = this.deps.executionRepository?.createProviderInvocationUsage(usageInput);
+        if (invocation && execInvocationId) {
+          this.deps.executionRepository?.updateExecutionInvocation(execInvocationId, {
+            providerInvocationId: invocation.id,
+          });
+        }
       }
 
-      if (invocation && execInvocationId) {
-        this.deps.executionRepository?.updateExecutionInvocation(execInvocationId, {
-          providerInvocationId: invocation.id,
-        });
+      if (execInvocationId) {
+        this.assertExecutionInvocationCanRun(execInvocationId);
       }
-
       const startedMs = Date.now();
       this.deps.logger?.info("Provider invocation started", {
         logPurpose: "invocation",
@@ -483,7 +494,7 @@ export class ProviderExecutionService {
             usageSignature !== lastPersistedUsageSignature
             && invocation
             && this.deps.executionRepository
-            && this.isProviderInvocationStillRunning(invocation.id)
+            && this.isProviderWorkStillRunning(invocation.id, execInvocationId)
           ) {
             const durationMs = Date.now() - startedMs;
             this.deps.executionRepository.updateProviderInvocationUsage(invocation.id, {
@@ -559,7 +570,7 @@ export class ProviderExecutionService {
             invocation
             && this.deps.executionRepository
             && !preserveForStartupRecovery
-            && this.isProviderInvocationStillRunning(invocation.id)
+            && this.isProviderWorkStillRunning(invocation.id, execInvocationId)
           ) {
             const finishedAt = new Date().toISOString();
             const durationMs = Date.now() - startedMs;
@@ -597,10 +608,16 @@ export class ProviderExecutionService {
       // Persist any buffered streaming activity from the completed run before recording usage.
       activityCoalescer?.stop();
 
+      if (args.invocationId && execInvocationId && !this.isExecutionInvocationStillRunning(execInvocationId)) {
+        this.assertExecutionInvocationCanRun(execInvocationId);
+      }
+
       if (invocation && this.deps.executionRepository) {
         const finishedAt = new Date().toISOString();
         const durationMs = Date.now() - startedMs;
-        if (this.isProviderInvocationStillRunning(invocation.id) && !isServerShutdownAbort(args.signal)) {
+        const shouldPersistTerminalUsage = this.isProviderWorkStillRunning(invocation.id, execInvocationId)
+          && !isServerShutdownAbort(args.signal);
+        if (shouldPersistTerminalUsage) {
           this.deps.executionRepository.updateProviderInvocationUsage(invocation.id, {
             status: (args.signal?.aborted || isRuntimeShutdownInProgress()) ? "cancelled" : (result.ok ? "completed" : "failed"),
             model: effectiveModel,
@@ -625,8 +642,8 @@ export class ProviderExecutionService {
           });
         }
 
-        if (args.taskRunId) {
-            this.deps.executionRepository.appendTaskRunEvent(args.taskRunId, "cli_provider_usage_reported", "system", {
+        if (args.taskRunId && shouldPersistTerminalUsage) {
+          this.deps.executionRepository.appendTaskRunEvent(args.taskRunId, "cli_provider_usage_reported", "system", {
             provider: args.provider,
             model: effectiveModel,
             purpose: args.purpose,
@@ -758,7 +775,7 @@ export class ProviderExecutionService {
         && !(retryDecision.kind === "rate_limit" && rateLimitRetryCount >= args.workflowSettings.maxRateLimitRetries)
         ? retryDecision.retryAtIso
         : null;
-      if (execInvocationId) {
+      if (execInvocationId && this.isExecutionInvocationStillRunning(execInvocationId)) {
         this.deps.executionRepository?.updateExecutionInvocation(execInvocationId, {
           lastErrorCategory: classification.category,
           lastErrorMessage: persistedUserMessage,
@@ -796,7 +813,7 @@ export class ProviderExecutionService {
             });
           }
 
-          if (execInvocationId) {
+          if (execInvocationId && this.isExecutionInvocationStillRunning(execInvocationId)) {
             this.deps.executionRepository?.appendExecutionInvocationMessage(execInvocationId, {
               role: "system",
               contentMarkdown: retryMessage,
@@ -893,6 +910,22 @@ export class ProviderExecutionService {
     return !current || current.status === "running";
   }
 
+  private isProviderWorkStillRunning(providerInvocationId: string, executionInvocationId: string | null): boolean {
+    return this.isProviderInvocationStillRunning(providerInvocationId)
+      && (!executionInvocationId || !this.isExecutionInvocationCancelled(executionInvocationId));
+  }
+
+  private assertExecutionInvocationCanRun(executionInvocationId: string): void {
+    const current = this.deps.executionRepository?.getExecutionInvocation?.(executionInvocationId);
+    if (current?.status === "cancelled") {
+      throw new Error(`Execution invocation ${executionInvocationId} is ${current.status}; provider execution will not continue.`);
+    }
+  }
+
+  private isExecutionInvocationCancelled(executionInvocationId: string): boolean {
+    return this.deps.executionRepository?.getExecutionInvocation?.(executionInvocationId)?.status === "cancelled";
+  }
+
   private isExecutionInvocationStillRunning(executionInvocationId: string): boolean {
     const current = this.deps.executionRepository?.getExecutionInvocation?.(executionInvocationId);
     return !current || current.status === "running" || current.status === "paused";

tests/backend/services/cli-workflow-service.test.ts

@@ -141,6 +141,10 @@ describe("CliWorkflowService unpushed commit detection", () => {
         storedInvocation = { ...input, id: "xi-preparation" };
         return storedInvocation;
       }),
+      updateExecutionInvocation: vi.fn().mockImplementation((_id: string, input: Record<string, unknown>) => {
+        Object.assign(storedInvocation!, input);
+        return storedInvocation;
+      }),
       appendExecutionInvocationMessage: vi.fn().mockImplementation(() => {
         callOrder.push("persist_message");
       }),
@@ -191,7 +195,14 @@ describe("CliWorkflowService unpushed commit detection", () => {
     vi.mocked(executeProviderStage).mockResolvedValue(buildProviderStageResult(
       "No repository changes were required.\nCODE_UX_TASK_OUTCOME: completed",
     ));
-    vi.mocked(executeGitFinalizeStage).mockResolvedValue({ hasChanges: false, committedChanges: false });
+    vi.mocked(executeGitFinalizeStage).mockImplementation(async () => {
+      expect(storedInvocation).toMatchObject({ status: "running" });
+      return { hasChanges: true, committedChanges: true, pushedBranch: "worker-1" };
+    });
+    vi.mocked(executePrFinalizeStage).mockImplementation(async () => {
+      expect(storedInvocation).toMatchObject({ status: "running" });
+      return { prUrl: "https://example.test/pull/1" };
+    });
     vi.mocked(executeCleanupStage).mockResolvedValue({ cleanedUp: false });
 
     const workflow = (service as any).runTaskWorkflow({
@@ -254,6 +265,7 @@ describe("CliWorkflowService unpushed commit detection", () => {
       expect.objectContaining({ executionInvocationId: "xi-preparation" }),
       "mock prompt",
     );
+    expect(storedInvocation).toMatchObject({ status: "completed" });
   });
 
   it("runs task workflow pipeline and handles error", async () => {

tests/backend/services/cli-workflow/pipeline/pipeline-stages.test.ts

@@ -431,6 +431,44 @@ describe("executePrepareStage", () => {
 });
 
 describe("executeProviderStage", () => {
+  it("reuses the preparation invocation and defers its completion past provider execution", async () => {
+    const ctx = createMockContext();
+    const executionInvocation = {
+      id: "exec-prepared",
+      status: "running",
+      providerInvocationId: null as string | null,
+    };
+    ctx.executionInvocationId = executionInvocation.id;
+    ctx.deps.executionRepository!.getExecutionInvocation = vi.fn().mockReturnValue(executionInvocation as any);
+    vi.mocked(ctx.deps.executionRepository!.updateExecutionInvocation).mockImplementation((_id, input) => {
+      Object.assign(executionInvocation, input);
+      return executionInvocation as any;
+    });
+    vi.mocked(ctx.providerRunner.runProvider).mockResolvedValueOnce({
+      ok: true,
+      stdout: "success",
+      stderr: "",
+      usageTelemetry: { transcriptText: "success transcript" } as any,
+    });
+
+    await executeProviderStage(ctx, "prompt");
+
+    expect(ctx.deps.executionRepository!.createExecutionInvocation).not.toHaveBeenCalled();
+    expect(ctx.deps.executionRepository!.createProviderInvocationUsage).toHaveBeenCalledOnce();
+    expect(executionInvocation).toMatchObject({
+      status: "running",
+      providerInvocationId: "usage-1",
+    });
+    expect(ctx.deps.executionRepository!.updateExecutionInvocation).not.toHaveBeenCalledWith(
+      "exec-prepared",
+      expect.objectContaining({ status: "completed" }),
+    );
+    expect(ctx.deps.executionRepository!.appendExecutionInvocationMessage).toHaveBeenCalledWith("exec-prepared", {
+      role: "user",
+      contentMarkdown: "prompt",
+    });
+  });
+
   it("passes the narrow clarification gateway and worker identity to a task-coding provider run", async () => {
     const ctx = createMockContext();
     ctx.agentPresetId = "assigned-worker";
@@ -544,6 +582,7 @@ describe("executeProviderStage", () => {
       }),
       undefined,
       undefined,
+      "exec-1",
     );
   });