mirror of
https://github.com/BradGroux/veritas-kanban.git
synced 2026-08-28 02:44:59 +00:00
refactor: centralize attempt completion lifecycle
This commit is contained in:
Brad Groux
parent
c9db917422
commit
03d4e94eab
4 changed files with 636 additions and 331 deletions
|
|
@ -681,6 +681,14 @@ Callback and remote-session terminal sources are accepted only for OpenClaw.
|
|||
CLI process and SDK stream providers reject callback transport even when an
|
||||
attempt ID and manifest digest are known.
|
||||
|
||||
Terminal task mutation crosses one `AttemptLifecycleCoordinator` seam. The
|
||||
coordinator validates the persisted runtime, envelope, and optional launch
|
||||
manifest bindings; enforces active-attempt ownership; retries bounded task
|
||||
revision conflicts; updates current and historical attempt state together;
|
||||
and treats only the exact persisted idempotency key as a safe duplicate.
|
||||
Provider adapters and restart recovery prepare evidence but cannot implement a
|
||||
parallel terminal persistence path.
|
||||
|
||||
Provider summaries, evidence, artifacts, and verification claims are bounded,
|
||||
redacted, and stored as unverified provider evidence. Veritas independently
|
||||
captures Git HEAD, post-launch files and commits, task verification state,
|
||||
|
|
|
|||
299
server/src/__tests__/attempt-lifecycle-coordinator.test.ts
Normal file
299
server/src/__tests__/attempt-lifecycle-coordinator.test.ts
Normal file
|
|
@ -0,0 +1,299 @@
|
|||
import { describe, expect, it } from 'vitest';
|
||||
import type { Task, TaskAttempt, TaskEnvelope, UpdateTaskInput } from '@veritas-kanban/shared';
|
||||
import { providerRuntimeManifestFixture } from './fixtures/provider-runtime-manifest.js';
|
||||
import {
|
||||
AttemptLifecycleCoordinator,
|
||||
type AttemptLifecycleStore,
|
||||
} from '../services/attempt-lifecycle-coordinator.js';
|
||||
import { ProviderCompletionService } from '../services/provider-completion-service.js';
|
||||
import { TaskEnvelopeService } from '../services/task-envelope-service.js';
|
||||
|
||||
const completedAt = '2026-08-24T05:00:00.000Z';
|
||||
|
||||
function baseTask(): Task {
|
||||
return {
|
||||
id: 'task_lifecycle_completion',
|
||||
title: 'Persist completion through one lifecycle owner',
|
||||
description: 'Keep terminal attempt mutation behind the lifecycle coordinator.',
|
||||
type: 'code',
|
||||
status: 'in-progress',
|
||||
priority: 'high',
|
||||
project: 'veritas-kanban',
|
||||
created: '2026-08-24T04:00:00.000Z',
|
||||
updated: '2026-08-24T04:00:00.000Z',
|
||||
revision: 4,
|
||||
executionPolicy: { commitPolicy: 'allowed' },
|
||||
};
|
||||
}
|
||||
|
||||
async function taskEnvelope(task: Task): Promise<TaskEnvelope> {
|
||||
return new TaskEnvelopeService({
|
||||
captureLaunchBaseline: async (_worktreePath, capturedAt) => ({
|
||||
capturedAt,
|
||||
headSha: 'a'.repeat(40),
|
||||
dirty: false,
|
||||
files: [],
|
||||
}),
|
||||
captureCompletionEvidence: async () => ({
|
||||
capturedAt: completedAt,
|
||||
headSha: 'b'.repeat(40),
|
||||
changedFiles: [],
|
||||
commits: [],
|
||||
artifacts: [],
|
||||
verification: [],
|
||||
sideEffects: [],
|
||||
}),
|
||||
}).build({
|
||||
task,
|
||||
attemptId: 'attempt_lifecycle_completion',
|
||||
createdAt: '2026-08-24T04:30:00.000Z',
|
||||
worktreePath: '/tmp/veritas-attempt-lifecycle',
|
||||
providerRuntimeManifest,
|
||||
commitPolicy: 'allowed',
|
||||
});
|
||||
}
|
||||
|
||||
class MemoryAttemptLifecycleStore implements AttemptLifecycleStore {
|
||||
constructor(private task: Task) {}
|
||||
|
||||
async getTask(taskId: string): Promise<Task | null> {
|
||||
return taskId === this.task.id ? structuredClone(this.task) : null;
|
||||
}
|
||||
|
||||
async updateTask(taskId: string, input: UpdateTaskInput): Promise<Task | null> {
|
||||
if (taskId !== this.task.id) return null;
|
||||
if (input.expectedRevision !== this.task.revision) {
|
||||
throw new Error('Task revision conflict');
|
||||
}
|
||||
const { expectedRevision: _expectedRevision, ...patch } = input;
|
||||
this.task = {
|
||||
...this.task,
|
||||
...patch,
|
||||
revision: (this.task.revision ?? 0) + 1,
|
||||
updated: completedAt,
|
||||
};
|
||||
return structuredClone(this.task);
|
||||
}
|
||||
}
|
||||
|
||||
async function completionFixture(summary = 'Lifecycle work completed.') {
|
||||
const task = baseTask();
|
||||
const providerRuntimeManifest = providerRuntimeManifestFixture();
|
||||
const envelope = await taskEnvelope(task);
|
||||
const attempt: TaskAttempt = {
|
||||
id: envelope.attempt.id,
|
||||
agent: 'codex',
|
||||
provider: envelope.launchManifest.provider,
|
||||
status: 'running',
|
||||
started: envelope.createdAt,
|
||||
providerRuntimeManifest,
|
||||
taskEnvelope: envelope,
|
||||
};
|
||||
const activeTask: Task = { ...task, attempt, attempts: [attempt] };
|
||||
const completionResult = await new ProviderCompletionService(
|
||||
{
|
||||
captureCompletionEvidence: async () => ({
|
||||
capturedAt: completedAt,
|
||||
headSha: 'b'.repeat(40),
|
||||
changedFiles: [],
|
||||
commits: [],
|
||||
artifacts: [],
|
||||
verification: [],
|
||||
sideEffects: [],
|
||||
}),
|
||||
},
|
||||
() => completedAt
|
||||
).complete({
|
||||
task: activeTask,
|
||||
taskEnvelope: envelope,
|
||||
claim: {
|
||||
terminalSource: 'process',
|
||||
status: 'success',
|
||||
summary,
|
||||
},
|
||||
});
|
||||
return { task: activeTask, attempt, completionResult };
|
||||
}
|
||||
|
||||
describe('AttemptLifecycleCoordinator', () => {
|
||||
it('persists terminal completion through the lifecycle seam', async () => {
|
||||
const { task, attempt, completionResult } = await completionFixture();
|
||||
const store = new MemoryAttemptLifecycleStore(task);
|
||||
const coordinator = new AttemptLifecycleCoordinator(store);
|
||||
|
||||
const outcome = await coordinator.persistCompletion({
|
||||
task,
|
||||
attempt,
|
||||
completionResult,
|
||||
});
|
||||
|
||||
expect(outcome.duplicate).toBe(false);
|
||||
expect(outcome.task).toMatchObject({
|
||||
status: 'done',
|
||||
revision: 5,
|
||||
attempt: {
|
||||
id: attempt.id,
|
||||
status: 'complete',
|
||||
ended: completedAt,
|
||||
completionResult: { idempotencyKey: completionResult.idempotencyKey },
|
||||
},
|
||||
});
|
||||
expect(outcome.task.attempts).toEqual([
|
||||
expect.objectContaining({ id: attempt.id, status: 'complete' }),
|
||||
]);
|
||||
await expect(store.getTask(task.id)).resolves.toEqual(outcome.task);
|
||||
});
|
||||
|
||||
it('retries a revision conflict against the same immutable attempt', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const historicalAttempt: TaskAttempt = {
|
||||
id: 'attempt_historical',
|
||||
agent: 'hermes',
|
||||
provider: 'hermes-cli',
|
||||
status: 'complete',
|
||||
ended: '2026-08-24T03:00:00.000Z',
|
||||
};
|
||||
const currentTask: Task = {
|
||||
...fixture.task,
|
||||
revision: 5,
|
||||
attempts: [historicalAttempt, fixture.attempt],
|
||||
};
|
||||
const coordinator = new AttemptLifecycleCoordinator(
|
||||
new MemoryAttemptLifecycleStore(currentTask)
|
||||
);
|
||||
|
||||
const outcome = await coordinator.persistCompletion(fixture);
|
||||
|
||||
expect(outcome.task.revision).toBe(6);
|
||||
expect(outcome.task.attempts?.map((attempt) => attempt.id)).toEqual([
|
||||
historicalAttempt.id,
|
||||
fixture.attempt.id,
|
||||
]);
|
||||
});
|
||||
|
||||
it('treats the same persisted terminal result as an idempotent duplicate', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const completedAttempt: TaskAttempt = {
|
||||
...fixture.attempt,
|
||||
status: 'complete',
|
||||
ended: completedAt,
|
||||
completionResult: fixture.completionResult,
|
||||
};
|
||||
const persistedTask: Task = {
|
||||
...fixture.task,
|
||||
status: 'done',
|
||||
revision: 5,
|
||||
attempt: completedAttempt,
|
||||
attempts: [completedAttempt],
|
||||
};
|
||||
const coordinator = new AttemptLifecycleCoordinator(
|
||||
new MemoryAttemptLifecycleStore(persistedTask)
|
||||
);
|
||||
|
||||
const outcome = await coordinator.persistCompletion(fixture);
|
||||
|
||||
expect(outcome.duplicate).toBe(true);
|
||||
expect(outcome.task).toEqual(persistedTask);
|
||||
});
|
||||
|
||||
it('fails closed when another attempt owns the task during retry', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const competingAttempt: TaskAttempt = {
|
||||
id: 'attempt_competing',
|
||||
agent: 'hermes',
|
||||
provider: 'hermes-cli',
|
||||
status: 'running',
|
||||
};
|
||||
const coordinator = new AttemptLifecycleCoordinator(
|
||||
new MemoryAttemptLifecycleStore({
|
||||
...fixture.task,
|
||||
revision: 5,
|
||||
attempt: competingAttempt,
|
||||
attempts: [fixture.attempt, competingAttempt],
|
||||
})
|
||||
);
|
||||
|
||||
await expect(coordinator.persistCompletion(fixture)).rejects.toMatchObject({
|
||||
statusCode: 409,
|
||||
code: 'CONFLICT',
|
||||
details: expect.objectContaining({
|
||||
activeAttemptId: competingAttempt.id,
|
||||
finalizationAttemptId: fixture.attempt.id,
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects stale completion input before the first persistence attempt', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const competingAttempt: TaskAttempt = {
|
||||
id: 'attempt_competing',
|
||||
agent: 'hermes',
|
||||
provider: 'hermes-cli',
|
||||
status: 'running',
|
||||
};
|
||||
const staleTask: Task = {
|
||||
...fixture.task,
|
||||
attempt: competingAttempt,
|
||||
attempts: [fixture.attempt, competingAttempt],
|
||||
};
|
||||
const coordinator = new AttemptLifecycleCoordinator(new MemoryAttemptLifecycleStore(staleTask));
|
||||
|
||||
await expect(
|
||||
coordinator.persistCompletion({ ...fixture, task: staleTask })
|
||||
).rejects.toMatchObject({
|
||||
statusCode: 409,
|
||||
code: 'CONFLICT',
|
||||
details: expect.objectContaining({
|
||||
activeAttemptId: competingAttempt.id,
|
||||
finalizationAttemptId: fixture.attempt.id,
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects a different persisted terminal result for the same attempt', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const competingFixture = await completionFixture('A different terminal claim completed.');
|
||||
const competingResult = competingFixture.completionResult;
|
||||
const completedAttempt: TaskAttempt = {
|
||||
...fixture.attempt,
|
||||
status: 'complete',
|
||||
ended: completedAt,
|
||||
completionResult: competingResult,
|
||||
};
|
||||
const persistedTask: Task = {
|
||||
...fixture.task,
|
||||
status: 'done',
|
||||
revision: 5,
|
||||
attempt: completedAttempt,
|
||||
attempts: [completedAttempt],
|
||||
};
|
||||
const coordinator = new AttemptLifecycleCoordinator(
|
||||
new MemoryAttemptLifecycleStore(persistedTask)
|
||||
);
|
||||
|
||||
await expect(coordinator.persistCompletion(fixture)).rejects.toMatchObject({
|
||||
statusCode: 409,
|
||||
code: 'CONFLICT',
|
||||
details: expect.objectContaining({
|
||||
persistedIdempotencyKey: competingResult.idempotencyKey,
|
||||
completionIdempotencyKey: fixture.completionResult.idempotencyKey,
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('preserves a non-active task status during startup reconciliation', async () => {
|
||||
const fixture = await completionFixture();
|
||||
const blockedTask: Task = { ...fixture.task, status: 'blocked' };
|
||||
const coordinator = new AttemptLifecycleCoordinator(
|
||||
new MemoryAttemptLifecycleStore(blockedTask)
|
||||
);
|
||||
|
||||
const outcome = await coordinator.persistCompletion({
|
||||
...fixture,
|
||||
task: blockedTask,
|
||||
preserveNonActiveTaskStatus: true,
|
||||
});
|
||||
|
||||
expect(outcome.task.status).toBe('blocked');
|
||||
});
|
||||
});
|
||||
289
server/src/services/attempt-lifecycle-coordinator.ts
Normal file
289
server/src/services/attempt-lifecycle-coordinator.ts
Normal file
|
|
@ -0,0 +1,289 @@
|
|||
import type {
|
||||
CompletionResult,
|
||||
Task,
|
||||
TaskAttempt,
|
||||
TaskCompletionStatus,
|
||||
TaskEnvelope,
|
||||
UpdateTaskInput,
|
||||
} from '@veritas-kanban/shared';
|
||||
import { ConflictError } from '../middleware/error-handler.js';
|
||||
import {
|
||||
parseCompletionResultForEnvelope,
|
||||
parseTaskEnvelope,
|
||||
} from '../schemas/task-envelope-schemas.js';
|
||||
import { parseRunLaunchManifest } from '../schemas/run-launch-manifest-schemas.js';
|
||||
import { assertProviderRuntimeManifestSnapshot } from './provider-runtime-control-service.js';
|
||||
|
||||
const COMPLETION_PERSISTENCE_ATTEMPTS = 3;
|
||||
|
||||
export interface AttemptLifecycleStore {
|
||||
getTask(taskId: string): Promise<Task | null>;
|
||||
updateTask(taskId: string, input: UpdateTaskInput): Promise<Task | null>;
|
||||
}
|
||||
|
||||
export interface PersistAttemptCompletionInput {
|
||||
task: Task;
|
||||
attempt: TaskAttempt;
|
||||
completionResult: CompletionResult;
|
||||
preserveNonActiveTaskStatus?: boolean;
|
||||
clearRunRecovery?: boolean;
|
||||
}
|
||||
|
||||
export interface PersistAttemptCompletionOutcome {
|
||||
task: Task;
|
||||
attempt: TaskAttempt;
|
||||
completionResult: CompletionResult;
|
||||
duplicate: boolean;
|
||||
}
|
||||
|
||||
export class CompletionOwnershipError extends ConflictError {}
|
||||
|
||||
/**
|
||||
* Owns terminal attempt mutation and its immutable persistence invariants.
|
||||
* Provider orchestration prepares completion evidence, then crosses this seam
|
||||
* exactly once to claim the terminal task state.
|
||||
*/
|
||||
export class AttemptLifecycleCoordinator {
|
||||
constructor(private readonly store: AttemptLifecycleStore) {}
|
||||
|
||||
parsePersistedCompletion(attempt: TaskAttempt): CompletionResult {
|
||||
if (!attempt.taskEnvelope || !attempt.completionResult) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted provider completion is missing its task envelope',
|
||||
{ attemptId: attempt.id }
|
||||
);
|
||||
}
|
||||
try {
|
||||
return parseCompletionResultForEnvelope(attempt.completionResult, attempt.taskEnvelope);
|
||||
} catch {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted provider completion failed integrity validation',
|
||||
{
|
||||
attemptId: attempt.id,
|
||||
remediation:
|
||||
'Repair or remove the corrupted completion record before accepting another terminal claim.',
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
assertCompletionBinding(taskId: string, attempt: TaskAttempt): TaskEnvelope {
|
||||
const providerRuntimeManifest = attempt.providerRuntimeManifest;
|
||||
const taskEnvelope = attempt.taskEnvelope;
|
||||
if (!providerRuntimeManifest || !taskEnvelope) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted attempt is missing immutable completion bindings',
|
||||
{ taskId, attemptId: attempt.id }
|
||||
);
|
||||
}
|
||||
|
||||
let parsedEnvelope: TaskEnvelope;
|
||||
let parsedRunLaunchManifest: ReturnType<typeof parseRunLaunchManifest> | undefined;
|
||||
try {
|
||||
assertProviderRuntimeManifestSnapshot(providerRuntimeManifest);
|
||||
parsedEnvelope = parseTaskEnvelope(taskEnvelope);
|
||||
parsedRunLaunchManifest = attempt.runLaunchManifest
|
||||
? parseRunLaunchManifest(attempt.runLaunchManifest)
|
||||
: undefined;
|
||||
} catch {
|
||||
throw new CompletionOwnershipError('Persisted attempt binding failed integrity validation', {
|
||||
taskId,
|
||||
attemptId: attempt.id,
|
||||
});
|
||||
}
|
||||
|
||||
const mismatches = [
|
||||
attempt.provider !== providerRuntimeManifest.provider && 'attempt runtime provider',
|
||||
parsedEnvelope.subject.id !== taskId && 'task ID',
|
||||
parsedEnvelope.attempt.id !== attempt.id && 'attempt ID',
|
||||
parsedEnvelope.launchManifest.digest !== providerRuntimeManifest.digest &&
|
||||
'envelope runtime digest',
|
||||
parsedEnvelope.launchManifest.provider !== providerRuntimeManifest.provider &&
|
||||
'envelope runtime provider',
|
||||
parsedEnvelope.launchManifest.adapter !== providerRuntimeManifest.adapter &&
|
||||
'envelope runtime adapter',
|
||||
parsedEnvelope.launchManifest.protocolVersion !== providerRuntimeManifest.protocolVersion &&
|
||||
'envelope runtime protocol',
|
||||
parsedRunLaunchManifest?.taskId !== undefined &&
|
||||
parsedRunLaunchManifest.taskId !== taskId &&
|
||||
'run launch task ID',
|
||||
parsedRunLaunchManifest?.attemptId !== undefined &&
|
||||
parsedRunLaunchManifest.attemptId !== attempt.id &&
|
||||
'run launch attempt ID',
|
||||
parsedRunLaunchManifest?.taskEnvelope.digest !== undefined &&
|
||||
parsedRunLaunchManifest.taskEnvelope.digest !== parsedEnvelope.digest &&
|
||||
'run launch task envelope digest',
|
||||
parsedRunLaunchManifest?.providerRuntime.digest !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.digest !== providerRuntimeManifest.digest &&
|
||||
'run launch runtime digest',
|
||||
parsedRunLaunchManifest?.providerRuntime.provider !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.provider !== providerRuntimeManifest.provider &&
|
||||
'run launch runtime provider',
|
||||
parsedRunLaunchManifest?.providerRuntime.adapter !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.adapter !== providerRuntimeManifest.adapter &&
|
||||
'run launch runtime adapter',
|
||||
].filter((field): field is string => typeof field === 'string');
|
||||
|
||||
if (mismatches.length > 0) {
|
||||
throw new CompletionOwnershipError('Persisted attempt completion bindings do not agree', {
|
||||
taskId,
|
||||
attemptId: attempt.id,
|
||||
mismatches,
|
||||
remediation: 'Repair the persisted attempt binding before accepting a terminal completion.',
|
||||
});
|
||||
}
|
||||
return parsedEnvelope;
|
||||
}
|
||||
|
||||
async persistCompletion(
|
||||
input: PersistAttemptCompletionInput
|
||||
): Promise<PersistAttemptCompletionOutcome> {
|
||||
const taskId = input.task.id;
|
||||
if (input.task.attempt?.id !== input.attempt.id) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Provider finalization does not match the active attempt',
|
||||
{
|
||||
taskId,
|
||||
activeAttemptId: input.task.attempt?.id,
|
||||
finalizationAttemptId: input.attempt.id,
|
||||
}
|
||||
);
|
||||
}
|
||||
const envelope = this.assertCompletionBinding(taskId, input.attempt);
|
||||
const completionResult = parseCompletionResultForEnvelope(input.completionResult, envelope);
|
||||
const completedAttempt: TaskAttempt = {
|
||||
...input.attempt,
|
||||
status: completionResult.status === 'success' ? 'complete' : 'failed',
|
||||
ended: completionResult.completedAt,
|
||||
completionResult,
|
||||
...(input.clearRunRecovery ? { runRecovery: undefined } : {}),
|
||||
};
|
||||
let taskSnapshot = input.task;
|
||||
let lastError: unknown;
|
||||
|
||||
for (
|
||||
let persistenceAttempt = 1;
|
||||
persistenceAttempt <= COMPLETION_PERSISTENCE_ATTEMPTS;
|
||||
persistenceAttempt++
|
||||
) {
|
||||
try {
|
||||
const statusUpdate =
|
||||
input.preserveNonActiveTaskStatus && taskSnapshot.status !== 'in-progress'
|
||||
? {}
|
||||
: { status: taskStatusForCompletion(completionResult.status) };
|
||||
const updatedTask = await this.store.updateTask(taskId, {
|
||||
expectedRevision: normalizedTaskRevision(taskSnapshot),
|
||||
...statusUpdate,
|
||||
attempt: completedAttempt,
|
||||
attempts: upsertAttemptHistory(taskSnapshot.attempts, completedAttempt),
|
||||
});
|
||||
if (!updatedTask) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Task was archived or deleted before completion could be persisted',
|
||||
{ taskId, attemptId: input.attempt.id }
|
||||
);
|
||||
}
|
||||
return {
|
||||
task: updatedTask,
|
||||
attempt: completedAttempt,
|
||||
completionResult,
|
||||
duplicate: false,
|
||||
};
|
||||
} catch (error) {
|
||||
if (error instanceof CompletionOwnershipError) throw error;
|
||||
lastError = error;
|
||||
|
||||
let latestTask: Task | null;
|
||||
try {
|
||||
latestTask = await this.store.getTask(taskId);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (!latestTask) continue;
|
||||
if (latestTask.attempt?.id !== input.attempt.id) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Provider finalization no longer matches the active attempt',
|
||||
{
|
||||
taskId,
|
||||
activeAttemptId: latestTask.attempt?.id,
|
||||
finalizationAttemptId: input.attempt.id,
|
||||
}
|
||||
);
|
||||
}
|
||||
this.assertRetryBinding(taskId, completedAttempt, latestTask.attempt);
|
||||
if (latestTask.attempt.completionResult) {
|
||||
const persisted = this.parsePersistedCompletion(latestTask.attempt);
|
||||
if (persisted.idempotencyKey === completionResult.idempotencyKey) {
|
||||
return {
|
||||
task: latestTask,
|
||||
attempt: latestTask.attempt,
|
||||
completionResult: persisted,
|
||||
duplicate: true,
|
||||
};
|
||||
}
|
||||
throw new CompletionOwnershipError(
|
||||
'A different terminal result already owns this attempt',
|
||||
{
|
||||
taskId,
|
||||
attemptId: input.attempt.id,
|
||||
persistedIdempotencyKey: persisted.idempotencyKey,
|
||||
completionIdempotencyKey: completionResult.idempotencyKey,
|
||||
}
|
||||
);
|
||||
}
|
||||
taskSnapshot = latestTask;
|
||||
}
|
||||
}
|
||||
|
||||
throw lastError instanceof Error
|
||||
? lastError
|
||||
: new Error('Provider completion persistence retry budget was exhausted');
|
||||
}
|
||||
|
||||
private assertRetryBinding(
|
||||
taskId: string,
|
||||
expectedAttempt: TaskAttempt,
|
||||
latestAttempt: TaskAttempt
|
||||
): void {
|
||||
this.assertCompletionBinding(taskId, latestAttempt);
|
||||
const mismatches = [
|
||||
expectedAttempt.provider !== latestAttempt.provider && 'provider',
|
||||
expectedAttempt.providerRuntimeManifest?.digest !==
|
||||
latestAttempt.providerRuntimeManifest?.digest && 'provider runtime manifest',
|
||||
expectedAttempt.taskEnvelope?.digest !== latestAttempt.taskEnvelope?.digest &&
|
||||
'task envelope',
|
||||
expectedAttempt.runLaunchManifest?.digest !== latestAttempt.runLaunchManifest?.digest &&
|
||||
'run launch manifest',
|
||||
].filter((field): field is string => typeof field === 'string');
|
||||
if (mismatches.length > 0) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted attempt binding changed during completion retry',
|
||||
{
|
||||
attemptId: latestAttempt.id,
|
||||
mismatches,
|
||||
remediation:
|
||||
'Discard the stale local finalizer and reconcile the currently persisted attempt.',
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function normalizedTaskRevision(task: Pick<Task, 'revision'>): number {
|
||||
return typeof task.revision === 'number' && Number.isInteger(task.revision) && task.revision >= 0
|
||||
? task.revision
|
||||
: 1;
|
||||
}
|
||||
|
||||
function taskStatusForCompletion(status: TaskCompletionStatus): 'done' | 'blocked' | 'in-progress' {
|
||||
if (status === 'success') return 'done';
|
||||
if (status === 'blocked') return 'blocked';
|
||||
return 'in-progress';
|
||||
}
|
||||
|
||||
function upsertAttemptHistory(
|
||||
history: TaskAttempt[] | undefined,
|
||||
attempt: TaskAttempt
|
||||
): TaskAttempt[] {
|
||||
return [...(history ?? []).filter((candidate) => candidate.id !== attempt.id), attempt];
|
||||
}
|
||||
|
|
@ -168,9 +168,7 @@ import {
|
|||
} from '../middleware/error-handler.js';
|
||||
import type { AgentBudgetThresholdEvent } from '@veritas-kanban/shared';
|
||||
import { getAgentProfilePackageService } from './agent-profile-package-service.js';
|
||||
import {
|
||||
ProviderRuntimeManifestService,
|
||||
} from './provider-runtime-manifest-service.js';
|
||||
import { ProviderRuntimeManifestService } from './provider-runtime-manifest-service.js';
|
||||
import type { WorkspaceFileRepository } from '../storage/interfaces.js';
|
||||
import { LocalWorkspaceFileRepository } from '../storage/workspace-file-repository.js';
|
||||
import {
|
||||
|
|
@ -197,11 +195,6 @@ import {
|
|||
} from './harness-support-profile-registry.js';
|
||||
import { RunLaunchManifestService, diffRunLaunchManifests } from './run-launch-manifest-service.js';
|
||||
import { RunLaunchCompiler } from './run-launch-compiler.js';
|
||||
import {
|
||||
parseCompletionResultForEnvelope,
|
||||
parseTaskEnvelope,
|
||||
} from '../schemas/task-envelope-schemas.js';
|
||||
import { parseRunLaunchManifest } from '../schemas/run-launch-manifest-schemas.js';
|
||||
import { RunTerminalExecuteRequestSchema } from '../schemas/run-terminal-schemas.js';
|
||||
import {
|
||||
ProviderCompletionService,
|
||||
|
|
@ -209,6 +202,10 @@ import {
|
|||
type ProviderCompletionEvidenceClaim,
|
||||
type ProviderTerminalClaim,
|
||||
} from './provider-completion-service.js';
|
||||
import {
|
||||
AttemptLifecycleCoordinator,
|
||||
CompletionOwnershipError,
|
||||
} from './attempt-lifecycle-coordinator.js';
|
||||
import { getCredentialBrokerService } from './credential-broker-service.js';
|
||||
import { WorktreeService } from './worktree-service.js';
|
||||
import {
|
||||
|
|
@ -609,7 +606,6 @@ const scheduledRecoveries = new Map<
|
|||
string,
|
||||
{ attemptId: string; timer: ReturnType<typeof setTimeout> }
|
||||
>();
|
||||
const COMPLETION_PERSISTENCE_ATTEMPTS = 3;
|
||||
const NOOP_CREDENTIAL_LEASE_LIFECYCLE: CredentialLeaseLifecycle = {
|
||||
async revokeRun() {
|
||||
return 0;
|
||||
|
|
@ -627,8 +623,6 @@ class CompletionPersistenceError extends Error {
|
|||
}
|
||||
}
|
||||
|
||||
class CompletionOwnershipError extends ConflictError {}
|
||||
|
||||
function normalizedTaskRevision(task: Pick<Task, 'revision'>): number {
|
||||
return typeof task.revision === 'number' && Number.isInteger(task.revision) && task.revision >= 0
|
||||
? task.revision
|
||||
|
|
@ -650,6 +644,7 @@ export class ClawdbotAgentService {
|
|||
private runLaunchManifests: RunLaunchManifestService;
|
||||
private runLaunchCompiler: RunLaunchCompiler;
|
||||
private providerCompletions: ProviderCompletionService;
|
||||
private attemptLifecycle: AttemptLifecycleCoordinator;
|
||||
private credentialLeases: CredentialLeaseLifecycle;
|
||||
private workspaceFiles: WorkspaceFileRepository;
|
||||
private worktrees: Pick<WorktreeService, 'claimOwnership' | 'releaseOwnership'>;
|
||||
|
|
@ -750,6 +745,7 @@ export class ClawdbotAgentService {
|
|||
this.taskEnvelopes = taskEnvelopes;
|
||||
this.runLaunchManifests = new RunLaunchManifestService();
|
||||
this.providerCompletions = providerCompletions;
|
||||
this.attemptLifecycle = new AttemptLifecycleCoordinator(this.taskService);
|
||||
this.credentialLeases = credentialLeases;
|
||||
this.workspaceFiles = workspaceFiles;
|
||||
this.worktrees =
|
||||
|
|
@ -1064,16 +1060,9 @@ export class ClawdbotAgentService {
|
|||
'Legacy running attempt has no durable supervisor bindings and cannot be recovered safely.',
|
||||
};
|
||||
if (attempt.taskEnvelope && attempt.providerRuntimeManifest) {
|
||||
await this.persistRestartedProviderCompletion(
|
||||
task,
|
||||
attempt,
|
||||
claim,
|
||||
this.providerCompletions.idempotencyKey({
|
||||
taskEnvelope: attempt.taskEnvelope,
|
||||
claim,
|
||||
}),
|
||||
{ preserveNonActiveTaskStatus: true }
|
||||
);
|
||||
await this.persistRestartedProviderCompletion(task, attempt, claim, {
|
||||
preserveNonActiveTaskStatus: true,
|
||||
});
|
||||
} else {
|
||||
const failedAttempt: TaskAttempt = {
|
||||
...attempt,
|
||||
|
|
@ -1090,7 +1079,7 @@ export class ClawdbotAgentService {
|
|||
continue;
|
||||
}
|
||||
|
||||
this.assertPersistedAttemptCompletionBinding(task.id, attempt);
|
||||
this.attemptLifecycle.assertCompletionBinding(task.id, attempt);
|
||||
const provider = executableProvider(attempt.provider);
|
||||
if (provider === 'system') {
|
||||
throw new CompletionOwnershipError('Persisted attempt has no executable provider.', {
|
||||
|
|
@ -3492,7 +3481,7 @@ export class ClawdbotAgentService {
|
|||
provenance.providerRuntimeManifestDigest &&
|
||||
persistedAttempt.taskEnvelope
|
||||
) {
|
||||
this.assertPersistedAttemptCompletionBinding(taskId, persistedAttempt);
|
||||
this.attemptLifecycle.assertCompletionBinding(taskId, persistedAttempt);
|
||||
this.assertTerminalTransport(persistedAttempt.provider, terminalSource);
|
||||
const claim = this.normalizeTerminalClaim(result, terminalSource);
|
||||
const idempotencyKey = this.providerCompletions.idempotencyKey({
|
||||
|
|
@ -3500,7 +3489,8 @@ export class ClawdbotAgentService {
|
|||
claim,
|
||||
});
|
||||
if (persistedAttempt.completionResult) {
|
||||
const persistedCompletion = this.parsePersistedCompletion(persistedAttempt);
|
||||
const persistedCompletion =
|
||||
this.attemptLifecycle.parsePersistedCompletion(persistedAttempt);
|
||||
if (persistedCompletion.idempotencyKey === idempotencyKey) {
|
||||
await this.admission.releaseByAttempt(
|
||||
persistedAttempt.taskEnvelope.workspace.workspaceId,
|
||||
|
|
@ -3537,12 +3527,7 @@ export class ClawdbotAgentService {
|
|||
(terminalSource === 'callback' || terminalSource === 'remote-session') &&
|
||||
(persistedAttempt.status === 'running' || persistedAttempt.status === 'failed')
|
||||
) {
|
||||
await this.persistRestartedProviderCompletion(
|
||||
task,
|
||||
persistedAttempt,
|
||||
claim,
|
||||
idempotencyKey
|
||||
);
|
||||
await this.persistRestartedProviderCompletion(task, persistedAttempt, claim);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
|
@ -3606,121 +3591,6 @@ export class ClawdbotAgentService {
|
|||
}
|
||||
}
|
||||
|
||||
private parsePersistedCompletion(attempt: TaskAttempt): CompletionResult {
|
||||
if (!attempt.taskEnvelope || !attempt.completionResult) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted provider completion is missing its task envelope',
|
||||
{
|
||||
attemptId: attempt.id,
|
||||
}
|
||||
);
|
||||
}
|
||||
try {
|
||||
return parseCompletionResultForEnvelope(attempt.completionResult, attempt.taskEnvelope);
|
||||
} catch {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted provider completion failed integrity validation',
|
||||
{
|
||||
attemptId: attempt.id,
|
||||
remediation:
|
||||
'Repair or remove the corrupted completion record before accepting another terminal claim.',
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private assertCompletionRetryBinding(
|
||||
taskId: string,
|
||||
expectedAttempt: TaskAttempt,
|
||||
latestAttempt: TaskAttempt
|
||||
): void {
|
||||
this.assertPersistedAttemptCompletionBinding(taskId, latestAttempt);
|
||||
const mismatches = [
|
||||
expectedAttempt.provider !== latestAttempt.provider && 'provider',
|
||||
expectedAttempt.providerRuntimeManifest?.digest !==
|
||||
latestAttempt.providerRuntimeManifest?.digest && 'provider runtime manifest',
|
||||
expectedAttempt.taskEnvelope?.digest !== latestAttempt.taskEnvelope?.digest &&
|
||||
'task envelope',
|
||||
expectedAttempt.runLaunchManifest?.digest !== latestAttempt.runLaunchManifest?.digest &&
|
||||
'run launch manifest',
|
||||
].filter((field): field is string => typeof field === 'string');
|
||||
if (mismatches.length > 0) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted attempt binding changed during completion retry',
|
||||
{
|
||||
attemptId: latestAttempt.id,
|
||||
mismatches,
|
||||
remediation:
|
||||
'Discard the stale local finalizer and reconcile the currently persisted attempt.',
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private assertPersistedAttemptCompletionBinding(taskId: string, attempt: TaskAttempt): void {
|
||||
const providerRuntimeManifest = attempt.providerRuntimeManifest;
|
||||
const taskEnvelope = attempt.taskEnvelope;
|
||||
if (!providerRuntimeManifest || !taskEnvelope) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Persisted attempt is missing immutable completion bindings',
|
||||
{ taskId, attemptId: attempt.id }
|
||||
);
|
||||
}
|
||||
let parsedEnvelope: TaskEnvelope;
|
||||
let parsedRunLaunchManifest: RunLaunchManifest | undefined;
|
||||
try {
|
||||
assertProviderRuntimeManifestSnapshot(providerRuntimeManifest);
|
||||
parsedEnvelope = parseTaskEnvelope(taskEnvelope);
|
||||
parsedRunLaunchManifest = attempt.runLaunchManifest
|
||||
? parseRunLaunchManifest(attempt.runLaunchManifest)
|
||||
: undefined;
|
||||
} catch {
|
||||
throw new CompletionOwnershipError('Persisted attempt binding failed integrity validation', {
|
||||
taskId,
|
||||
attemptId: attempt.id,
|
||||
});
|
||||
}
|
||||
const mismatches = [
|
||||
attempt.provider !== providerRuntimeManifest.provider && 'attempt runtime provider',
|
||||
parsedEnvelope.subject.id !== taskId && 'task ID',
|
||||
parsedEnvelope.attempt.id !== attempt.id && 'attempt ID',
|
||||
parsedEnvelope.launchManifest.digest !== providerRuntimeManifest.digest &&
|
||||
'envelope runtime digest',
|
||||
parsedEnvelope.launchManifest.provider !== providerRuntimeManifest.provider &&
|
||||
'envelope runtime provider',
|
||||
parsedEnvelope.launchManifest.adapter !== providerRuntimeManifest.adapter &&
|
||||
'envelope runtime adapter',
|
||||
parsedEnvelope.launchManifest.protocolVersion !== providerRuntimeManifest.protocolVersion &&
|
||||
'envelope runtime protocol',
|
||||
parsedRunLaunchManifest?.taskId !== undefined &&
|
||||
parsedRunLaunchManifest.taskId !== taskId &&
|
||||
'run launch task ID',
|
||||
parsedRunLaunchManifest?.attemptId !== undefined &&
|
||||
parsedRunLaunchManifest.attemptId !== attempt.id &&
|
||||
'run launch attempt ID',
|
||||
parsedRunLaunchManifest?.taskEnvelope.digest !== undefined &&
|
||||
parsedRunLaunchManifest.taskEnvelope.digest !== parsedEnvelope.digest &&
|
||||
'run launch task envelope digest',
|
||||
parsedRunLaunchManifest?.providerRuntime.digest !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.digest !== providerRuntimeManifest.digest &&
|
||||
'run launch runtime digest',
|
||||
parsedRunLaunchManifest?.providerRuntime.provider !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.provider !== providerRuntimeManifest.provider &&
|
||||
'run launch runtime provider',
|
||||
parsedRunLaunchManifest?.providerRuntime.adapter !== undefined &&
|
||||
parsedRunLaunchManifest.providerRuntime.adapter !== providerRuntimeManifest.adapter &&
|
||||
'run launch runtime adapter',
|
||||
].filter((field): field is string => typeof field === 'string');
|
||||
if (mismatches.length > 0) {
|
||||
throw new CompletionOwnershipError('Persisted attempt completion bindings do not agree', {
|
||||
taskId,
|
||||
attemptId: attempt.id,
|
||||
mismatches,
|
||||
remediation: 'Repair the persisted attempt binding before accepting a terminal completion.',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async persistSupervisorCompletion(
|
||||
task: Task,
|
||||
attempt: TaskAttempt,
|
||||
|
|
@ -3732,63 +3602,13 @@ export class ClawdbotAgentService {
|
|||
attemptId: attempt.id,
|
||||
});
|
||||
}
|
||||
this.assertPersistedAttemptCompletionBinding(task.id, attempt);
|
||||
const completionResult = parseCompletionResultForEnvelope(value, attempt.taskEnvelope);
|
||||
const completedAttempt: TaskAttempt = {
|
||||
...attempt,
|
||||
status: completionResult.status === 'success' ? 'complete' : 'failed',
|
||||
ended: completionResult.completedAt,
|
||||
completionResult,
|
||||
runRecovery: undefined,
|
||||
};
|
||||
let taskSnapshot = task;
|
||||
let lastError: unknown;
|
||||
for (
|
||||
let persistenceAttempt = 1;
|
||||
persistenceAttempt <= COMPLETION_PERSISTENCE_ATTEMPTS;
|
||||
persistenceAttempt++
|
||||
) {
|
||||
try {
|
||||
const updatedTask = await this.taskService.updateTask(task.id, {
|
||||
expectedRevision: normalizedTaskRevision(taskSnapshot),
|
||||
status: taskStatusForCompletion(completionResult.status),
|
||||
attempt: completedAttempt,
|
||||
attempts: upsertAttemptHistory(taskSnapshot.attempts, completedAttempt),
|
||||
});
|
||||
if (!updatedTask) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Task was archived or deleted before supervisor completion could be persisted.',
|
||||
{ taskId: task.id, attemptId: attempt.id }
|
||||
);
|
||||
}
|
||||
lastError = undefined;
|
||||
break;
|
||||
} catch (error) {
|
||||
if (error instanceof CompletionOwnershipError) throw error;
|
||||
lastError = error;
|
||||
const latestTask = await this.taskService.getTask(task.id);
|
||||
if (latestTask?.attempt?.id !== attempt.id) throw error;
|
||||
this.assertCompletionRetryBinding(task.id, completedAttempt, latestTask.attempt);
|
||||
if (latestTask.attempt.completionResult) {
|
||||
const persisted = this.parsePersistedCompletion(latestTask.attempt);
|
||||
if (persisted.idempotencyKey === completionResult.idempotencyKey) {
|
||||
lastError = undefined;
|
||||
break;
|
||||
}
|
||||
throw new CompletionOwnershipError(
|
||||
'A different terminal result already owns this attempt.',
|
||||
{
|
||||
taskId: task.id,
|
||||
attemptId: attempt.id,
|
||||
persistedIdempotencyKey: persisted.idempotencyKey,
|
||||
completionIdempotencyKey: completionResult.idempotencyKey,
|
||||
}
|
||||
);
|
||||
}
|
||||
taskSnapshot = latestTask;
|
||||
}
|
||||
}
|
||||
if (lastError) throw lastError;
|
||||
const persisted = await this.attemptLifecycle.persistCompletion({
|
||||
task,
|
||||
attempt,
|
||||
completionResult: value,
|
||||
clearRunRecovery: true,
|
||||
});
|
||||
const { completionResult, attempt: completedAttempt } = persisted;
|
||||
this.scheduleReflectionExtraction(attempt.taskEnvelope.workspace.workspaceId, completionResult);
|
||||
await this.admission.releaseByAttempt(
|
||||
attempt.taskEnvelope.workspace.workspaceId,
|
||||
|
|
@ -3824,7 +3644,6 @@ export class ClawdbotAgentService {
|
|||
task: Task,
|
||||
attempt: TaskAttempt,
|
||||
claim: ProviderTerminalClaim,
|
||||
idempotencyKey: string,
|
||||
options: { preserveNonActiveTaskStatus?: boolean } = {}
|
||||
): Promise<void> {
|
||||
if (!attempt.taskEnvelope) {
|
||||
|
|
@ -3833,7 +3652,7 @@ export class ClawdbotAgentService {
|
|||
attemptId: attempt.id,
|
||||
});
|
||||
}
|
||||
this.assertPersistedAttemptCompletionBinding(task.id, attempt);
|
||||
this.attemptLifecycle.assertCompletionBinding(task.id, attempt);
|
||||
const dependencyCircuits = attempt.runLaunchManifest
|
||||
? await this.captureDependencyCircuits(
|
||||
attempt.runLaunchManifest.providerRuntime.provider,
|
||||
|
|
@ -3871,13 +3690,6 @@ export class ClawdbotAgentService {
|
|||
task.id,
|
||||
attempt.id
|
||||
);
|
||||
const completedAttempt: TaskAttempt = {
|
||||
...attempt,
|
||||
status: completionResult.status === 'success' ? 'complete' : 'failed',
|
||||
ended: completionResult.completedAt,
|
||||
completionResult,
|
||||
};
|
||||
const completionStatus = taskStatusForCompletion(completionResult.status);
|
||||
if (attempt.provider === 'openclaw' && completionResult.summary) {
|
||||
await this.appendMappedProviderEvent(
|
||||
task,
|
||||
|
|
@ -3952,58 +3764,12 @@ export class ClawdbotAgentService {
|
|||
completionResult
|
||||
);
|
||||
}
|
||||
let taskSnapshot = task;
|
||||
let lastError: unknown;
|
||||
for (
|
||||
let persistenceAttempt = 1;
|
||||
persistenceAttempt <= COMPLETION_PERSISTENCE_ATTEMPTS;
|
||||
persistenceAttempt++
|
||||
) {
|
||||
const taskStatusUpdate =
|
||||
options.preserveNonActiveTaskStatus && taskSnapshot.status !== 'in-progress'
|
||||
? {}
|
||||
: { status: completionStatus };
|
||||
try {
|
||||
const updatedTask = await this.taskService.updateTask(task.id, {
|
||||
expectedRevision: normalizedTaskRevision(taskSnapshot),
|
||||
...taskStatusUpdate,
|
||||
attempt: completedAttempt,
|
||||
attempts: upsertAttemptHistory(taskSnapshot.attempts, completedAttempt),
|
||||
});
|
||||
if (!updatedTask) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Task was archived or deleted before completion could be persisted',
|
||||
{ taskId: task.id, attemptId: attempt.id }
|
||||
);
|
||||
}
|
||||
lastError = undefined;
|
||||
break;
|
||||
} catch (error) {
|
||||
if (error instanceof CompletionOwnershipError) throw error;
|
||||
lastError = error;
|
||||
const latestTask = await this.taskService.getTask(task.id);
|
||||
if (latestTask?.attempt?.id !== attempt.id) throw error;
|
||||
this.assertCompletionRetryBinding(task.id, completedAttempt, latestTask.attempt);
|
||||
if (latestTask.attempt.completionResult) {
|
||||
const latestCompletion = this.parsePersistedCompletion(latestTask.attempt);
|
||||
if (latestCompletion.idempotencyKey === idempotencyKey) {
|
||||
lastError = undefined;
|
||||
break;
|
||||
}
|
||||
throw new CompletionOwnershipError(
|
||||
'A different terminal result already owns this attempt',
|
||||
{
|
||||
taskId: task.id,
|
||||
attemptId: attempt.id,
|
||||
persistedIdempotencyKey: latestCompletion.idempotencyKey,
|
||||
completionIdempotencyKey: idempotencyKey,
|
||||
}
|
||||
);
|
||||
}
|
||||
taskSnapshot = latestTask;
|
||||
}
|
||||
}
|
||||
if (lastError) throw lastError;
|
||||
await this.attemptLifecycle.persistCompletion({
|
||||
task,
|
||||
attempt,
|
||||
completionResult,
|
||||
preserveNonActiveTaskStatus: options.preserveNonActiveTaskStatus,
|
||||
});
|
||||
this.scheduleReflectionExtraction(
|
||||
attempt.taskEnvelope.workspace.workspaceId,
|
||||
completionResult,
|
||||
|
|
@ -4557,69 +4323,18 @@ export class ClawdbotAgentService {
|
|||
pending: PendingAgent,
|
||||
prepared: NonNullable<PendingAgent['preparedCompletion']>
|
||||
): Promise<boolean> {
|
||||
let taskSnapshot = prepared.taskBeforeCompletion;
|
||||
let lastError: unknown;
|
||||
for (let attempt = 1; attempt <= COMPLETION_PERSISTENCE_ATTEMPTS; attempt++) {
|
||||
if (!taskSnapshot) {
|
||||
throw new ConflictError('Task disappeared before completion could be persisted', {
|
||||
taskId,
|
||||
attemptId: pending.attemptId,
|
||||
});
|
||||
}
|
||||
try {
|
||||
const updatedTask = await this.taskService.updateTask(taskId, {
|
||||
expectedRevision: normalizedTaskRevision(taskSnapshot),
|
||||
status: taskStatusForCompletion(prepared.completionResult.status),
|
||||
attempt: prepared.completedAttempt,
|
||||
attempts: upsertAttemptHistory(taskSnapshot.attempts, prepared.completedAttempt),
|
||||
});
|
||||
if (!updatedTask) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Task was archived or deleted before completion could be persisted',
|
||||
{ taskId, attemptId: pending.attemptId }
|
||||
);
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error instanceof CompletionOwnershipError) throw error;
|
||||
lastError = error;
|
||||
let latestTask: Task | null;
|
||||
try {
|
||||
latestTask = await this.taskService.getTask(taskId);
|
||||
} catch {
|
||||
latestTask = null;
|
||||
}
|
||||
if (!latestTask) continue;
|
||||
if (latestTask.attempt?.id !== pending.attemptId) {
|
||||
throw new CompletionOwnershipError(
|
||||
'Provider finalization no longer matches the active attempt',
|
||||
{
|
||||
taskId,
|
||||
activeAttemptId: latestTask.attempt?.id,
|
||||
finalizationAttemptId: pending.attemptId,
|
||||
}
|
||||
);
|
||||
}
|
||||
this.assertCompletionRetryBinding(taskId, prepared.completedAttempt, latestTask.attempt);
|
||||
if (latestTask.attempt.completionResult) {
|
||||
const persisted = this.parsePersistedCompletion(latestTask.attempt);
|
||||
if (persisted.idempotencyKey === prepared.completionResult.idempotencyKey) return true;
|
||||
throw new CompletionOwnershipError(
|
||||
'A different terminal result already owns this attempt',
|
||||
{
|
||||
taskId,
|
||||
attemptId: pending.attemptId,
|
||||
persistedIdempotencyKey: persisted.idempotencyKey,
|
||||
completionIdempotencyKey: prepared.completionResult.idempotencyKey,
|
||||
}
|
||||
);
|
||||
}
|
||||
taskSnapshot = latestTask;
|
||||
}
|
||||
if (!prepared.taskBeforeCompletion) {
|
||||
throw new ConflictError('Task disappeared before completion could be persisted', {
|
||||
taskId,
|
||||
attemptId: pending.attemptId,
|
||||
});
|
||||
}
|
||||
throw lastError instanceof Error
|
||||
? lastError
|
||||
: new Error('Provider completion persistence retry budget was exhausted');
|
||||
await this.attemptLifecycle.persistCompletion({
|
||||
task: prepared.taskBeforeCompletion,
|
||||
attempt: prepared.completedAttempt,
|
||||
completionResult: prepared.completionResult,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -10560,12 +10275,6 @@ ${message}
|
|||
`;
|
||||
}
|
||||
|
||||
function taskStatusForCompletion(status: TaskCompletionStatus): 'done' | 'blocked' | 'in-progress' {
|
||||
if (status === 'success') return 'done';
|
||||
if (status === 'blocked') return 'blocked';
|
||||
return 'in-progress';
|
||||
}
|
||||
|
||||
function admissionReleaseReason(
|
||||
status: TaskCompletionStatus | undefined,
|
||||
success?: boolean
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue