diff --git a/docs/AGENT-PROVIDERS.md b/docs/AGENT-PROVIDERS.md index 8bcec8f0..cda61e6c 100644 --- a/docs/AGENT-PROVIDERS.md +++ b/docs/AGENT-PROVIDERS.md @@ -973,13 +973,14 @@ Configure ceilings through `PATCH /api/settings/features`: ``` An individual request larger than a ceiling receives a terminal policy denial. -Temporary exhaustion returns a retryable overload decision with the limiting -scope and bounded retry guidance. A fresh direct task start with no transient -operator message or per-request policy override is instead persisted as one -bounded `admission-queue-entry/v1` record. The start response has -`status: "queued"`, `queueId`, the reserved `attemptId`, `retryAfterMs`, and -redacted limiting scopes. Harnesses must treat that response as accepted work, -not as a failed attempt, and must not submit a duplicate start. +Queueable temporary exhaustion persists one bounded +`admission-queue-entry/v1` record. A fresh direct task start with no transient +operator message or per-request policy override returns `status: "queued"`, +`queueId`, the reserved `attemptId`, `retryAfterMs`, and redacted limiting +scopes. Harnesses must treat that response as accepted work, not as a failed +attempt, and must not submit a duplicate start. Workflow roots and +provider-backed workflow steps instead persist `state: "waiting"` in their +admission binding while the run and step remain `pending`. The queue uses deterministic FIFO order. A worker claims the queue lease and admission capacity atomically, then reruns provider, sandbox, budget, workspace @@ -988,17 +989,21 @@ supervisor ownership changes the entry to `dispatched`; after that point only run recovery may restart or finish the work. Abandoned pre-dispatch leases requeue with bounded backoff. Terminal drift, retry exhaustion, and queue overflow fail closed. Queue records retain task, workspace, agent, execution -tree, and limiting-scope evidence, but never prompts, messages, tool arguments, -credentials, or raw idempotency keys. +tree, workflow version and revision, retry or fallback sequence, provider, +host, immutable runtime and phase digests, and limiting-scope evidence. They +never retain prompts, workflow context, messages, tool arguments, credentials, +or raw idempotency keys. Active leases renew while the verified run is live; completion, interruption, cancellation, or launch failure releases the reservation idempotently. Workflow retry and fallback attempts release the prior step reservation before -acquiring another. After restart, task runs use their durable supervisor and -workflow runs use the exact persisted root binding before reclaiming a -reservation. Unverified in-flight workflow steps are released and blocked for -operator reconciliation. Caller-supplied idempotency values are represented by -a stable SHA-256 identity in durable records; the original value is not stored. +queueing or acquiring the replacement. After restart, task runs use their +durable supervisor and workflow runs use the exact persisted root and step +bindings before reclaiming a reservation. Roots and steps interrupted after +durable queue dispatch but before provider execution resume exactly once. +Unverified provider work that was already running remains blocked for operator +reconciliation. Caller-supplied idempotency values are represented by a stable +SHA-256 identity in durable records; the original value is not stored. Every reservation also carries a versioned execution-tree identity. Resume, follow-up, fork, retry, fallback, provider handoff, workflow-step, and diff --git a/docs/API-REFERENCE.md b/docs/API-REFERENCE.md index 280136da..4064ce1a 100644 --- a/docs/API-REFERENCE.md +++ b/docs/API-REFERENCE.md @@ -4177,9 +4177,10 @@ credentials or prompts. Workflow records identify the run, optional step, and root reservation. Overloaded starts return a `409` conflict with `details.code` set to `ADMISSION_OVERLOAD`, the limiting scope, and bounded `retryAfterMs`. A request larger than a configured ceiling returns -`ADMISSION_POLICY_DENIED`. A step-level retryable overload blocks the workflow -without creating a partially active attempt; terminal policy denial fails the -run and releases its root. +`ADMISSION_POLICY_DENIED`. A queueable workflow-root or provider-step overload +instead persists the run or step with admission state `waiting` and status +`pending`, without creating a partially active provider attempt. Terminal +policy denial fails the run and releases its root. Execution-tree reservations include `execution-tree-identity/v1` and durable requested, remaining, committed, released-unused, and idempotent usage-event @@ -4219,6 +4220,21 @@ with `ADMISSION_OVERLOAD`. Queue bounds and retry behavior are configured under `features.admission.queue`. Queue inspection endpoints are added separately; reservation inspection remains unchanged. +Workflow roots and provider-backed steps use the same durable queue internally. +Their queue targets preserve the workflow version, run revision, task and step +identity, retry or fallback sequence, execution-tree edge, selected provider +and host, and immutable runtime and phase digests. Queue records never copy +workflow context, prompts, tool arguments, or credentials. Before dispatch, +Veritas revalidates the persisted run, workflow version, step state, provider +runtime, host, phase authority, budgets, and reservation evidence. Drift +terminalizes the queue entry and run without calling the provider. + +A pre-dispatch failure releases capacity and may requeue with bounded backoff. +Once the queue entry is durably `dispatched`, workflow recovery owns the run. +Startup reconciliation resumes a root or step that stopped at that ownership +boundary exactly once; already-running provider work continues to use the +existing operator-reconciliation rules. + --- ## Traces diff --git a/docs/API-WORKFLOWS.md b/docs/API-WORKFLOWS.md index d93be6e6..53ce6ae1 100644 --- a/docs/API-WORKFLOWS.md +++ b/docs/API-WORKFLOWS.md @@ -2141,13 +2141,14 @@ export interface ParallelSubStep { ```typescript export type WorkflowRunStatus = 'pending' | 'running' | 'blocked' | 'completed' | 'failed'; +export type WorkflowAdmissionState = 'waiting' | 'dispatching' | 'active' | 'terminal'; export interface WorkflowRun { id: string; // run__ workflowId: string; workflowVersion: number; taskId?: string; // optional task association - admission?: WorkflowRootAdmissionBinding; // durable root reservation identity + admission?: WorkflowRootAdmissionBinding; // durable root reservation or queue identity status: WorkflowRunStatus; currentStep?: string; // current step ID context: Record; // shared context across steps @@ -2176,7 +2177,7 @@ export interface StepRun { retries: number; output?: string; // path to output file error?: string; - admission?: WorkflowStepAdmissionBinding; // latest executable attempt decision + admission?: WorkflowStepAdmissionBinding; // latest executable attempt or queue decision // Loop-specific state loopState?: { @@ -2188,6 +2189,12 @@ export interface StepRun { } ``` +When root or step capacity is temporarily unavailable, the corresponding +admission binding has `state: "waiting"` and a durable `queueEntryId`. The run +and step remain `pending`; provider execution is not marked active. A claimed +entry briefly uses `dispatching` while Veritas transfers durable ownership to +workflow recovery, then becomes `active` before provider execution. + ### ToolPolicy ```typescript diff --git a/docs/WORKFLOW_ENGINE_ARCHITECTURE.md b/docs/WORKFLOW_ENGINE_ARCHITECTURE.md index 126830bb..ad135e47 100644 --- a/docs/WORKFLOW_ENGINE_ARCHITECTURE.md +++ b/docs/WORKFLOW_ENGINE_ARCHITECTURE.md @@ -2472,10 +2472,13 @@ transaction. Each step commits only its own provider-reported usage; cumulative workflow totals are not copied into child records. This keeps deep, wide, and replayed workflows attributable without descendant double counting. -Temporary step overload blocks the workflow with the limiting policy retained -in the run; an impossible request fails the run. Durable fair queues and -priority aging are scheduler concerns layered on this controller, not hidden -inside the workflow service. +Temporary root or step overload persists one bounded FIFO admission entry and +leaves the run or step visibly waiting without provider dispatch. Claims +revalidate workflow, run, provider, host, phase, budget, and execution-tree +evidence before transferring ownership to workflow recovery. Retry and +fallback replacements cannot queue until predecessor capacity is released. An +impossible request or stale queue authority fails closed. Priority aging and +cross-workspace fairness remain later scheduler concerns. **OpenClaw Session Cleanup**: diff --git a/server/src/__tests__/admission-control-service.test.ts b/server/src/__tests__/admission-control-service.test.ts index 125cf173..05fdba24 100644 --- a/server/src/__tests__/admission-control-service.test.ts +++ b/server/src/__tests__/admission-control-service.test.ts @@ -205,6 +205,31 @@ describe('AdmissionControlService', () => { await expect(repository.list({ taskId: 'task-recover' })).resolves.toHaveLength(1); }); + it('does not resurrect an intentionally released run during restart recovery', async () => { + const repository = await repositoryFor('file'); + const service = createService(repository, configuredSettings()); + const decision = await service.admit(request('task-released-recovery')); + const bound = await service.bindAttempt( + decision.reservation?.id as string, + 'attempt-released-recovery' + ); + await service.release(bound.id, 'completed', 'release-before-recovery'); + + await expect( + service.recoverVerifiedRun({ + workspaceId: 'workspace-a', + taskId: 'task-released-recovery', + attemptId: 'attempt-released-recovery', + }) + ).rejects.toMatchObject({ + statusCode: 409, + details: { + reservationId: bound.id, + reservationState: 'released', + }, + }); + }); + it('does not release another attempt when an idempotent retry loses the bind race', async () => { const repository = await repositoryFor('file'); const settings = configuredSettings(); @@ -483,6 +508,57 @@ describe.each(['file', 'sqlite'] as const)('%s admission reservation parity', (b await expect(recovered.claimNextQueued()).resolves.toBeNull(); }); + it('reactivates a released pre-dispatch reservation when the queue retries', async () => { + const repository = await repositoryFor(backend); + let now = new Date('2026-07-25T12:00:00.000Z'); + const settings = configuredSettings({ + global: { concurrentRuns: 1 }, + queue: { retryBackoffMs: 250 }, + }); + const service = createService(repository, settings, { + ownerId: `owner-retry-${backend}`, + now: () => now, + }); + const active = await service.admit(request(`task-retry-active-${backend}`)); + const queued = await service.admitOrQueue(request(`task-retry-queued-${backend}`), { + agent: 'codex', + attemptId: `attempt-retry-${backend}`, + }); + await service.release( + active.reservation?.id as string, + 'completed', + `release-retry-active-${backend}` + ); + const firstClaim = await service.claimNextQueued(); + await service.release( + firstClaim?.reservation.id as string, + 'start-failed', + `release-retry-claim-${backend}` + ); + await service.requeueQueueEntry( + queued.queueEntry?.id as string, + 'TRANSIENT_PRE_DISPATCH', + 'Transient pre-dispatch failure.' + ); + now = new Date('2026-07-25T12:00:00.300Z'); + + const secondClaim = await service.claimNextQueued(); + expect(secondClaim).toMatchObject({ + entry: { + id: queued.queueEntry?.id, + state: 'leased', + retryCount: 1, + }, + reservation: { + id: firstClaim?.reservation.id, + state: 'active', + }, + }); + expect(secondClaim?.reservation.revision).toBeGreaterThan( + firstClaim?.reservation.revision as number + ); + }); + it('atomically reserves, converts, summarizes, and releases execution-tree budgets', async () => { const repository = await repositoryFor(backend); const service = createService(repository, configuredSettings()); diff --git a/server/src/__tests__/helpers/workflow-admission-stub.ts b/server/src/__tests__/helpers/workflow-admission-stub.ts index 5d77c118..e74ff0a0 100644 --- a/server/src/__tests__/helpers/workflow-admission-stub.ts +++ b/server/src/__tests__/helpers/workflow-admission-stub.ts @@ -7,23 +7,25 @@ export function workflowAdmissionStub(): AdmissionControlService { { budgetPolicies?: import('@veritas-kanban/shared').ExecutionTreeBudgetPolicy[] } >(); let pendingReservationId = 'admission_workflow_test'; + const admit = async (input: { + workflowRunId?: string; + workflowStepId?: string; + budgetPolicies?: import('@veritas-kanban/shared').ExecutionTreeBudgetPolicy[]; + }) => { + pendingReservationId = `admission_${input.workflowRunId ?? 'run'}_${input.workflowStepId ?? 'root'}`; + requestsByReservation.set(pendingReservationId, { + budgetPolicies: input.budgetPolicies, + }); + return { + outcome: 'admitted' as const, + reservation: { id: pendingReservationId }, + }; + }; return { getExecutionHostId: () => 'test-execution-host', - admit: async (input: { - workflowRunId?: string; - workflowStepId?: string; - budgetPolicies?: import('@veritas-kanban/shared').ExecutionTreeBudgetPolicy[]; - }) => { - pendingReservationId = `admission_${input.workflowRunId ?? 'run'}_${input.workflowStepId ?? 'root'}`; - requestsByReservation.set(pendingReservationId, { - budgetPolicies: input.budgetPolicies, - }); - return { - outcome: 'admitted', - reservation: { id: pendingReservationId }, - }; - }, + admit, + admitOrQueue: admit, get: async (id: string) => ({ id, request: requestsByReservation.get(id) ?? {} }), recordBudgetUsage: async (id: string) => ({ id }), bindAttempt: async (id: string, attemptId: string) => { diff --git a/server/src/__tests__/workflow-admission.test.ts b/server/src/__tests__/workflow-admission.test.ts index a291c305..04cfb6cd 100644 --- a/server/src/__tests__/workflow-admission.test.ts +++ b/server/src/__tests__/workflow-admission.test.ts @@ -74,9 +74,19 @@ function preparation(step: WorkflowDefinition['steps'][number]): WorkflowAgentSt kind: 'agent', step, runtimeProvider: 'codex-sdk', + runtimeManifest: { + digest: `sha256:${'a'.repeat(64)}`, + }, + requiredRuntimeCapabilities: [], hostRouting: { selectedHostId: 'local-process', }, + phaseAuthority: { + evidence: { + digest: `sha256:${'b'.repeat(64)}`, + }, + }, + phaseLaunchDigest: `sha256:${'c'.repeat(64)}`, } as WorkflowAgentStepPreparation; } @@ -130,11 +140,44 @@ async function createHarness( } as never, stepExecutor, admission, + requestAdmissionQueueDrain: () => undefined, }); runs.push(service); return { root, database, repository, admission, definition, service }; } +function restartHarness( + harness: Awaited>, + admissionSettings: AdmissionSettings, + stepExecutor: WorkflowStepExecutor, + ownerId: string +) { + harness.service.dispose(); + harness.admission.dispose(); + const admission = new AdmissionControlService({ + repository: harness.repository, + settings: async () => structuredClone(admissionSettings), + hostId: 'workflow-host', + ownerId, + processId: 303, + }); + admissions.push(admission); + const service = new WorkflowRunService({ + runsDir: path.join(harness.root, 'runs'), + storageType: harness.database ? 'sqlite' : 'file', + sqliteDatabase: harness.database, + workflowService: { + loadWorkflow: async (id: string) => + id === harness.definition.id ? harness.definition : null, + } as never, + stepExecutor, + admission, + requestAdmissionQueueDrain: () => undefined, + }); + runs.push(service); + return { admission, service }; +} + describe('workflow admission', () => { it.each(['file', 'sqlite'] as const)( 'reserves roots and provider steps atomically with %s storage', @@ -218,6 +261,484 @@ describe('workflow admission', () => { } ); + it.each(['file', 'sqlite'] as const)( + 'queues a saturated workflow root and resumes it exactly once with %s storage', + async (backend) => { + const executeStep = vi.fn(async (step: WorkflowDefinition['steps'][number]) => ({ + output: { completed: true }, + outputPath: `/tmp/${step.id}.json`, + })); + const harness = await createHarness( + backend, + settings({ global: { concurrentRuns: 2 } }), + executor(executeStep) + ); + const blockers = await Promise.all( + ['one', 'two'].map((suffix) => + harness.admission.admit({ + taskId: `root-blocker-${suffix}`, + workspaceId: 'other-workspace', + provider: 'workflow-control', + hostId: 'workflow-host', + idempotencyKey: `root-blocker-${backend}-${suffix}`, + requested: { runSlots: 1, processSlots: 0, estimatedMemoryMb: 0 }, + }) + ) + ); + expect(blockers.every((decision) => decision.outcome === 'admitted')).toBe(true); + + const rawContextSecret = `workflow-root-secret-${backend}`; + const run = await harness.service.startRun(harness.definition.id, undefined, { + operatorPrompt: `Use ${rawContextSecret}`, + toolArguments: { credential: rawContextSecret }, + }); + expect(run).toMatchObject({ + status: 'pending', + admission: { + state: 'waiting', + decision: { outcome: 'queued' }, + }, + }); + expect(executeStep).not.toHaveBeenCalled(); + + const [queued] = await harness.admission.listQueue({ + taskId: `workflow-root:${run.id}`, + }); + expect(queued).toMatchObject({ + state: 'queued', + target: { + kind: 'workflow-root', + workflowId: harness.definition.id, + workflowVersion: harness.definition.version, + workflowRunId: run.id, + workflowRunRevision: run.revision, + }, + }); + expect(JSON.stringify(queued)).not.toContain(rawContextSecret); + + await Promise.all( + blockers.map((decision, index) => + harness.admission.release( + decision.reservation?.id as string, + 'completed', + `release-root-blocker-${backend}-${index}` + ) + ) + ); + const claim = await harness.admission.claimNextQueued(); + expect(claim?.entry.id).toBe(queued?.id); + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + + await vi.waitFor(async () => { + expect((await harness.service.getRun(run.id))?.status).toBe('completed'); + }); + expect(executeStep).toHaveBeenCalledTimes(1); + await expect(harness.admission.getQueueEntry(queued?.id as string)).resolves.toMatchObject({ + state: 'dispatched', + dispatchedAttemptId: `workflow-root:${run.id}`, + }); + } + ); + + it.each(['file', 'sqlite'] as const)( + 'recovers a durably dispatched workflow root exactly once after restart with %s storage', + async (backend) => { + const executeStep = vi.fn(async (step: WorkflowDefinition['steps'][number]) => ({ + output: { completed: true }, + outputPath: `/tmp/${step.id}.json`, + })); + const admissionSettings = settings({ global: { concurrentRuns: 2 } }); + const stepExecutor = executor(executeStep); + const harness = await createHarness(backend, admissionSettings, stepExecutor); + const blockers = await Promise.all( + ['one', 'two'].map((suffix) => + harness.admission.admit({ + taskId: `restart-root-blocker-${suffix}`, + workspaceId: 'other-workspace', + provider: 'workflow-control', + hostId: 'workflow-host', + idempotencyKey: `restart-root-blocker-${backend}-${suffix}`, + requested: { runSlots: 1, processSlots: 0, estimatedMemoryMb: 0 }, + }) + ) + ); + const run = await harness.service.startRun(harness.definition.id); + await Promise.all( + blockers.map((decision, index) => + harness.admission.release( + decision.reservation?.id as string, + 'completed', + `release-restart-root-blocker-${backend}-${index}` + ) + ) + ); + const claim = await harness.admission.claimNextQueued(); + const markQueueDispatched = harness.admission.markQueueDispatched.bind(harness.admission); + vi.spyOn(harness.admission, 'markQueueDispatched').mockImplementationOnce( + async (queueId, attemptId) => { + await markQueueDispatched(queueId, attemptId); + throw new Error('simulated process exit after durable queue dispatch'); + } + ); + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + await expect(harness.service.getRun(run.id)).resolves.toMatchObject({ + status: 'pending', + admission: { state: 'dispatching' }, + }); + expect(executeStep).not.toHaveBeenCalled(); + + const restarted = restartHarness( + harness, + admissionSettings, + stepExecutor, + `owner-restarted-root-${backend}` + ); + await restarted.service.reconcilePendingRecoveries(); + + await vi.waitFor(async () => { + expect((await restarted.service.getRun(run.id))?.status).toBe('completed'); + }); + expect(executeStep).toHaveBeenCalledTimes(1); + await vi.waitFor(async () => { + expect( + (await restarted.admission.list({ workflowRunId: run.id })).every( + (reservation) => reservation.state === 'released' + ) + ).toBe(true); + }); + } + ); + + it.each(['file', 'sqlite'] as const)( + 'queues a saturated workflow step and dispatches it exactly once with %s storage', + async (backend) => { + const executeStep = vi.fn(async (step: WorkflowDefinition['steps'][number]) => ({ + output: { completed: true }, + outputPath: `/tmp/${step.id}.json`, + })); + const harness = await createHarness( + backend, + settings({ global: { concurrentRuns: 3 } }), + executor(executeStep) + ); + const blocker = await harness.admission.admit({ + taskId: 'step-provider-blocker', + workspaceId: 'other-workspace', + provider: 'codex-sdk', + hostId: 'local-process', + idempotencyKey: `step-provider-blocker-${backend}`, + requested: { runSlots: 1, processSlots: 1, estimatedMemoryMb: 0 }, + }); + expect(blocker.outcome).toBe('admitted'); + + const started = await harness.service.startRun(harness.definition.id); + const waiting = await vi.waitFor(async () => { + const run = await harness.service.getRun(started.id); + expect(run).toMatchObject({ + status: 'pending', + steps: [ + { + stepId: 'execute', + status: 'pending', + admission: { + state: 'waiting', + decision: { outcome: 'queued' }, + }, + }, + ], + }); + if (!run) throw new Error('Expected a persisted workflow run'); + return run; + }); + expect(executeStep).not.toHaveBeenCalled(); + + const [queued] = (await harness.admission.listQueue({})).filter( + (entry) => entry.target?.kind === 'workflow-step' + ); + expect(queued).toMatchObject({ + state: 'queued', + target: { + kind: 'workflow-step', + workflowId: harness.definition.id, + workflowVersion: harness.definition.version, + workflowRunId: waiting.id, + workflowRunRevision: waiting.revision, + workflowStepId: 'execute', + workflowStepSequence: 1, + recoverySequence: 0, + provider: 'codex-sdk', + hostId: 'local-process', + providerRuntimeManifestDigest: `sha256:${'a'.repeat(64)}`, + phaseEvidenceDigest: `sha256:${'b'.repeat(64)}`, + phaseLaunchDigest: `sha256:${'c'.repeat(64)}`, + }, + }); + expect(JSON.stringify(queued)).not.toContain('Run the test.'); + + await harness.admission.release( + blocker.reservation?.id as string, + 'completed', + `release-step-blocker-${backend}` + ); + const claim = await harness.admission.claimNextQueued(); + expect(claim?.entry.id).toBe(queued?.id); + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + + await vi.waitFor(async () => { + expect((await harness.service.getRun(waiting.id))?.status).toBe('completed'); + }); + expect(executeStep).toHaveBeenCalledTimes(1); + await expect(harness.admission.getQueueEntry(queued?.id as string)).resolves.toMatchObject({ + state: 'dispatched', + }); + } + ); + + it.each(['file', 'sqlite'] as const)( + 'recovers a durably dispatched workflow step exactly once after restart with %s storage', + async (backend) => { + const executeStep = vi.fn(async (step: WorkflowDefinition['steps'][number]) => ({ + output: { completed: true }, + outputPath: `/tmp/${step.id}.json`, + })); + const admissionSettings = settings({ global: { concurrentRuns: 3 } }); + const stepExecutor = executor(executeStep); + const harness = await createHarness(backend, admissionSettings, stepExecutor); + const blocker = await harness.admission.admit({ + taskId: 'restart-step-provider-blocker', + workspaceId: 'other-workspace', + provider: 'codex-sdk', + hostId: 'local-process', + idempotencyKey: `restart-step-provider-blocker-${backend}`, + requested: { runSlots: 1, processSlots: 1, estimatedMemoryMb: 0 }, + }); + const run = await harness.service.startRun(harness.definition.id); + await vi.waitFor(async () => { + expect((await harness.service.getRun(run.id))?.steps[0]?.admission?.state).toBe('waiting'); + }); + await harness.admission.release( + blocker.reservation?.id as string, + 'completed', + `release-restart-step-blocker-${backend}` + ); + const claim = await harness.admission.claimNextQueued(); + const markQueueDispatched = harness.admission.markQueueDispatched.bind(harness.admission); + vi.spyOn(harness.admission, 'markQueueDispatched').mockImplementationOnce( + async (queueId, attemptId) => { + await markQueueDispatched(queueId, attemptId); + throw new Error('simulated process exit after durable step dispatch'); + } + ); + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + await expect(harness.service.getRun(run.id)).resolves.toMatchObject({ + status: 'pending', + steps: [ + expect.objectContaining({ + stepId: 'execute', + admission: expect.objectContaining({ state: 'dispatching' }), + }), + ], + }); + expect(executeStep).not.toHaveBeenCalled(); + + const restarted = restartHarness( + harness, + admissionSettings, + stepExecutor, + `owner-restarted-step-${backend}` + ); + await restarted.service.reconcilePendingRecoveries(); + + await vi.waitFor(async () => { + expect((await restarted.service.getRun(run.id))?.status).toBe('completed'); + }); + expect(executeStep).toHaveBeenCalledTimes(1); + await vi.waitFor(async () => { + expect( + (await restarted.admission.list({ workflowRunId: run.id })).every( + (reservation) => reservation.state === 'released' + ) + ).toBe(true); + }); + } + ); + + it.each(['retry', 'fallback'] as const)( + 'queues a saturated %s replacement only after releasing its predecessor', + async (replacement) => { + let invocationCount = 0; + const executeStep = vi.fn(async (step: WorkflowDefinition['steps'][number]) => { + invocationCount += 1; + if (invocationCount === 1) throw new Error('ECONNRESET before replacement'); + return { output: { completed: true }, outputPath: `/tmp/${step.id}.json` }; + }); + const harness = await createHarness('file', settings(), executor(executeStep)); + harness.definition.steps[0].on_fail = + replacement === 'retry' + ? { retry: 1, retry_delay_ms: 250 } + : { retry: 0, retry_delay_ms: 250, escalate_to: 'agent:agent' }; + + const run = await harness.service.startRun(harness.definition.id); + await vi.waitFor(async () => { + expect((await harness.service.getRun(run.id))?.steps[0]?.runRetry?.state).toBe('scheduled'); + }); + const blocker = await harness.admission.admit({ + taskId: `${replacement}-replacement-blocker`, + workspaceId: 'other-workspace', + provider: 'codex-sdk', + hostId: 'local-process', + idempotencyKey: `${replacement}-replacement-blocker`, + requested: { runSlots: 1, processSlots: 1, estimatedMemoryMb: 0 }, + }); + expect(blocker.outcome).toBe('admitted'); + + const waiting = await vi.waitFor( + async () => { + const persisted = await harness.service.getRun(run.id); + expect(persisted?.steps[0]?.admission?.state).toBe('waiting'); + if (!persisted) throw new Error('Expected replacement workflow run'); + return persisted; + }, + { timeout: 2_000 } + ); + const reservations = await harness.admission.list({ workflowRunId: run.id }); + const predecessor = reservations.find( + (reservation) => + reservation.request.workflowStepId === 'execute' && + reservation.attemptId !== waiting.steps[0]?.admission?.attemptId + ); + expect(predecessor).toMatchObject({ + state: 'released', + release: { reason: 'failed' }, + }); + const [queued] = (await harness.admission.listQueue({})).filter( + (entry) => entry.target?.kind === 'workflow-step' + ); + expect(queued).toMatchObject({ + state: 'queued', + request: { + source: replacement === 'retry' ? 'recovery' : 'fallback', + budgetRequest: { retries: 1 }, + }, + target: { + kind: 'workflow-step', + workflowStepSequence: 2, + recoverySequence: 1, + edge: replacement, + }, + }); + + await harness.admission.release( + blocker.reservation?.id as string, + 'completed', + `release-${replacement}-replacement-blocker` + ); + const claim = await harness.admission.claimNextQueued(); + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + + await vi.waitFor(async () => { + expect((await harness.service.getRun(run.id))?.status).toBe('completed'); + }); + expect(executeStep).toHaveBeenCalledTimes(2); + await vi.waitFor(async () => { + expect( + (await harness.admission.list({ workflowRunId: run.id })).every( + (reservation) => reservation.state === 'released' + ) + ).toBe(true); + }); + } + ); + + it('terminalizes a queued step without provider dispatch when launch authority drifts', async () => { + let runtimeManifestDigest = `sha256:${'a'.repeat(64)}`; + const executeStep = vi.fn(async () => ({ + output: { completed: true }, + outputPath: '/tmp/execute.json', + })); + const stepExecutor = executor(executeStep); + stepExecutor.prepareStep = async (step) => ({ + ...preparation(step), + runtimeManifest: { digest: runtimeManifestDigest }, + }); + const harness = await createHarness( + 'file', + settings({ global: { concurrentRuns: 3 } }), + stepExecutor + ); + const blocker = await harness.admission.admit({ + taskId: 'drift-provider-blocker', + workspaceId: 'other-workspace', + provider: 'codex-sdk', + hostId: 'local-process', + idempotencyKey: 'drift-provider-blocker', + requested: { runSlots: 1, processSlots: 1, estimatedMemoryMb: 0 }, + }); + const started = await harness.service.startRun(harness.definition.id); + await vi.waitFor(async () => { + expect((await harness.service.getRun(started.id))?.steps[0]?.admission?.state).toBe( + 'waiting' + ); + }); + const [queued] = (await harness.admission.listQueue({})).filter( + (entry) => entry.target?.kind === 'workflow-step' + ); + await harness.admission.release( + blocker.reservation?.id as string, + 'completed', + 'release-drift-blocker' + ); + const claim = await harness.admission.claimNextQueued(); + runtimeManifestDigest = `sha256:${'d'.repeat(64)}`; + + await ( + harness.service as unknown as { + dispatchQueuedAdmission: (input: NonNullable) => Promise; + } + ).dispatchQueuedAdmission(claim as NonNullable); + + await expect(harness.admission.getQueueEntry(queued?.id as string)).resolves.toMatchObject({ + state: 'terminal', + terminal: { code: 'WORKFLOW_QUEUE_AUTHORITY_DRIFT' }, + }); + await expect(harness.service.getRun(started.id)).resolves.toMatchObject({ + status: 'failed', + steps: [ + expect.objectContaining({ + stepId: 'execute', + status: 'failed', + admission: expect.objectContaining({ state: 'terminal' }), + }), + ], + }); + expect(executeStep).not.toHaveBeenCalled(); + expect( + (await harness.admission.list({ workflowRunId: started.id })).map( + (reservation) => reservation.state + ) + ).toEqual(['released', 'released']); + }); + it('fails closed before provider dispatch when a step exceeds its provider ceiling', async () => { const executeStep = vi.fn(async () => ({ output: { completed: true }, diff --git a/server/src/schemas/admission-control-schemas.ts b/server/src/schemas/admission-control-schemas.ts index b26aa2f9..f5d77d0a 100644 --- a/server/src/schemas/admission-control-schemas.ts +++ b/server/src/schemas/admission-control-schemas.ts @@ -18,6 +18,7 @@ import { AgentBudgetLimitsSchema, AgentBudgetUsageSchema } from './agent-budget- const identifier = z.string().trim().min(1).max(240); const policyIdentifier = z.string().trim().min(1).max(256); +const digest = z.string().regex(/^sha256:[a-f0-9]{64}$/); const admissionProvider = z.enum([...EXECUTABLE_AGENT_PROVIDERS, ADMISSION_CONTROL_PROVIDER]); const executionTreeBudgetPolicyScope = z.enum([ 'workspace', @@ -213,6 +214,48 @@ export const AdmissionReservationSchema = z } }); +export const AdmissionQueueTargetSchema = z.discriminatedUnion('kind', [ + z + .object({ + kind: z.literal('direct'), + agent: identifier, + }) + .strict(), + z + .object({ + kind: z.literal('workflow-root'), + workflowId: identifier, + workflowVersion: z.number().int().positive(), + workflowRunId: identifier, + workflowRunRevision: z.number().int().positive(), + associatedTaskId: identifier.optional(), + initialContextDigest: digest, + budgetPolicyDigest: digest, + executionTreeDigest: digest, + }) + .strict(), + z + .object({ + kind: z.literal('workflow-step'), + workflowId: identifier, + workflowVersion: z.number().int().positive(), + workflowRunId: identifier, + workflowRunRevision: z.number().int().positive(), + workflowStepId: identifier, + workflowStepSequence: z.number().int().positive(), + recoverySequence: z.number().int().nonnegative(), + parentNodeId: identifier, + edge: z.enum(EXECUTION_TREE_EDGE_KINDS), + provider: z.enum(EXECUTABLE_AGENT_PROVIDERS), + hostId: identifier, + providerRuntimeManifestDigest: digest, + requiredRuntimeCapabilitiesDigest: digest, + phaseEvidenceDigest: digest, + phaseLaunchDigest: digest, + }) + .strict(), +]); + export const AdmissionQueueEntrySchema = z .object({ schemaVersion: z.literal(ADMISSION_QUEUE_ENTRY_SCHEMA_VERSION), @@ -220,7 +263,8 @@ export const AdmissionQueueEntrySchema = z revision: z.number().int().positive(), state: z.enum(ADMISSION_QUEUE_STATES), enqueueSequence: z.number().int().positive(), - agent: identifier, + agent: identifier.optional(), + target: AdmissionQueueTargetSchema.optional(), attemptId: identifier, request: AdmissionRequestSchema, policies: z.array(AdmissionLimitPolicySchema).max(6), @@ -246,6 +290,52 @@ export const AdmissionQueueEntrySchema = z }) .strict() .superRefine((entry, ctx) => { + const directAgent = entry.target?.kind === 'direct' ? entry.target.agent : entry.agent; + if (!entry.target && !entry.agent) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Admission queue entries require a launch target.', + path: ['target'], + }); + } + if (entry.target?.kind === 'direct' && entry.agent && entry.agent !== directAgent) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Direct queue target and legacy agent identity must match.', + path: ['agent'], + }); + } + if (entry.target && entry.target.kind !== 'direct' && entry.agent) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Workflow queue targets cannot contain a direct agent identity.', + path: ['agent'], + }); + } + if ( + entry.target?.kind === 'workflow-root' && + (entry.request.workflowRunId !== entry.target.workflowRunId || + entry.request.workflowStepId !== undefined) + ) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Workflow root queue target must match its admission request.', + path: ['target'], + }); + } + if ( + entry.target?.kind === 'workflow-step' && + (entry.request.workflowRunId !== entry.target.workflowRunId || + entry.request.workflowStepId !== entry.target.workflowStepId || + entry.request.provider !== entry.target.provider || + entry.request.hostId !== entry.target.hostId) + ) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Workflow step queue target must match its admission request.', + path: ['target'], + }); + } if (entry.state === 'leased' && (!entry.lease || !entry.reservationId)) { ctx.addIssue({ code: z.ZodIssueCode.custom, diff --git a/server/src/services/admission-control-service.ts b/server/src/services/admission-control-service.ts index 3d9685aa..5613bfa3 100644 --- a/server/src/services/admission-control-service.ts +++ b/server/src/services/admission-control-service.ts @@ -9,6 +9,7 @@ import type { AdmissionQueueClaim, AdmissionQueueEntry, AdmissionQueueListQuery, + AdmissionQueueTarget, AdmissionReservation, AdmissionReservationClaimOrQueueResult, AdmissionReservationListQuery, @@ -69,6 +70,13 @@ export interface DirectAdmissionQueueInput { attemptId: string; } +export interface WorkflowAdmissionQueueInput { + target: Exclude; + attemptId: string; +} + +export type AdmissionQueueInput = DirectAdmissionQueueInput | WorkflowAdmissionQueueInput; + export interface RecoverAdmissionInput { workspaceId: string; taskId: string; @@ -162,14 +170,14 @@ export class AdmissionControlService { async admitOrQueue( input: AdmissionRequestInput, - queue: DirectAdmissionQueueInput + queue: AdmissionQueueInput ): Promise { return this.admitInternal(input, queue); } private async admitInternal( input: AdmissionRequestInput, - queue?: DirectAdmissionQueueInput + queue?: AdmissionQueueInput ): Promise { const settings = await this.settings(); const now = this.now(); @@ -237,13 +245,33 @@ export class AdmissionControlService { createdAt: now.toISOString(), updatedAt: now.toISOString(), }); - const queueable = Boolean(queue && settings.queue.enabled && request.source === 'direct'); + const queueTarget: AdmissionQueueTarget | null = + queue && 'target' in queue + ? queue.target + : queue + ? { kind: 'direct', agent: queue.agent } + : null; + const queueable = Boolean( + queue && + queueTarget && + settings.queue.enabled && + ((queueTarget.kind === 'direct' && request.source === 'direct') || + (queueTarget.kind === 'workflow-root' && + request.source === 'workflow' && + request.workflowRunId === queueTarget.workflowRunId && + !request.workflowStepId) || + (queueTarget.kind === 'workflow-step' && + ['workflow', 'recovery', 'fallback'].includes(request.source) && + request.workflowRunId === queueTarget.workflowRunId && + request.workflowStepId === queueTarget.workflowStepId)) + ); const claimed: AdmissionReservationClaimOrQueueResult = queueable ? await this.repository.claimOrEnqueue({ record, queue: { id: queueEntryId(idempotencyKey), - agent: queue?.agent as AgentType, + ...(queueTarget?.kind === 'direct' ? { agent: queueTarget.agent } : {}), + target: queueTarget as AdmissionQueueTarget, attemptId: queue?.attemptId as string, request, policies, diff --git a/server/src/services/clawdbot-agent-service.ts b/server/src/services/clawdbot-agent-service.ts index 2075bbec..160067a7 100644 --- a/server/src/services/clawdbot-agent-service.ts +++ b/server/src/services/clawdbot-agent-service.ts @@ -672,6 +672,23 @@ export class ClawdbotAgentService { const claim = await this.admission.claimNextQueued(); if (!claim) return false; const { entry, reservation } = claim; + if (entry.target?.kind === 'workflow-root' || entry.target?.kind === 'workflow-step') { + const { getWorkflowRunService } = await import('./workflow-run-service.js'); + await getWorkflowRunService().dispatchQueuedAdmission(claim); + continue; + } + const queuedAgent = entry.target?.kind === 'direct' ? entry.target.agent : entry.agent; + if (!queuedAgent) { + await this.admission + .release(reservation.id, 'start-failed', `queue-target-missing:${entry.id}`) + .catch(() => {}); + await this.admission.terminateQueueEntry( + entry.id, + 'ADMISSION_QUEUE_TARGET_MISSING', + 'The queued direct launch has no agent target.' + ); + continue; + } if (startingAgents.has(entry.request.taskId) || pendingAgents.has(entry.request.taskId)) { await this.admission .release(reservation.id, 'start-failed', `queue-task-busy:${entry.id}`) @@ -686,7 +703,7 @@ export class ClawdbotAgentService { startingAgents.add(entry.request.taskId); try { - const result = await this.startReservedAgent(entry.request.taskId, entry.agent, { + const result = await this.startReservedAgent(entry.request.taskId, queuedAgent, { rootTaskId: entry.request.rootTaskId, admissionQueueClaim: claim, }); @@ -2390,7 +2407,11 @@ export class ClawdbotAgentService { const driftFields = [ queuedClaim.entry.state !== 'leased' ? 'queueState' : undefined, queuedClaim.entry.attemptId !== attemptId ? 'attemptId' : undefined, - queuedClaim.entry.agent !== agent ? 'agent' : undefined, + (queuedClaim.entry.target?.kind === 'direct' + ? queuedClaim.entry.target.agent + : queuedClaim.entry.agent) !== agent + ? 'agent' + : undefined, queuedClaim.entry.reservationId !== queuedClaim.reservation.id ? 'reservationId' : undefined, diff --git a/server/src/services/workflow-run-service.ts b/server/src/services/workflow-run-service.ts index 0af87203..727c0cbb 100644 --- a/server/src/services/workflow-run-service.ts +++ b/server/src/services/workflow-run-service.ts @@ -5,7 +5,7 @@ import fs from 'fs/promises'; import path from 'path'; -import { createHash, randomUUID } from 'node:crypto'; +import { createHash } from 'node:crypto'; import { nanoid } from 'nanoid'; import { ADMISSION_CONTROL_PROVIDER, @@ -18,6 +18,8 @@ import { type AgentBudgetUsage, type AgentType, type AdmissionDecision, + type AdmissionQueueClaim, + type AdmissionQueueEntry, type AdmissionReservation, type AdmissionReservationRelease, type ExecutionTreeBudgetPolicy, @@ -55,6 +57,7 @@ import { getAdmissionControlService, } from './admission-control-service.js'; import { normalizeWorkspaceId } from './task-envelope-service.js'; +import { digestRunLaunchValue } from '../utils/run-launch-manifest-digest.js'; const log = createLogger('workflow-run'); @@ -113,6 +116,7 @@ export class WorkflowRunService { private readonly ownsSqliteDatabase: boolean = false; private readonly runRecoveryPolicy: RunRecoveryPolicyService; private readonly admission: AdmissionControlService; + private readonly requestAdmissionQueueDrain?: () => void; constructor(options: string | WorkflowRunServiceOptions = {}) { const resolvedOptions = typeof options === 'string' ? { runsDir: options } : options; @@ -120,6 +124,7 @@ export class WorkflowRunService { this.workflowService = resolvedOptions.workflowService ?? getWorkflowService(); this.runRecoveryPolicy = resolvedOptions.runRecoveryPolicy ?? new RunRecoveryPolicyService(); this.admission = resolvedOptions.admission ?? getAdmissionControlService(); + this.requestAdmissionQueueDrain = resolvedOptions.requestAdmissionQueueDrain; this.stepExecutor = resolvedOptions.stepExecutor ?? new WorkflowStepExecutor(resolvedOptions.runsDir, { @@ -424,7 +429,7 @@ export class WorkflowRunService { 'Root objective budget' ), ]); - const decision = await this.admission.admit({ + const admissionInput = { taskId: admissionTaskId, rootTaskId, workspaceId, @@ -432,7 +437,7 @@ export class WorkflowRunService { hostId: this.admission.getExecutionHostId(), source: 'workflow', workflowRunId: run.id, - idempotencyKey: `workflow-root:${run.id}:${randomUUID()}`, + idempotencyKey: `workflow-root:${run.id}`, requested: { runSlots: 1, processSlots: 0, @@ -441,7 +446,34 @@ export class WorkflowRunService { executionTree, budgetPolicies, budgetRequest: { fanOut: 1 }, + } as const; + const decision = await this.admission.admitOrQueue(admissionInput, { + attemptId, + target: { + kind: 'workflow-root', + workflowId: run.workflowId, + workflowVersion: run.workflowVersion, + workflowRunId: run.id, + workflowRunRevision: (run.revision ?? 0) + 1, + ...(run.taskId ? { associatedTaskId: run.taskId } : {}), + initialContextDigest: digestRunLaunchValue(run.context), + budgetPolicyDigest: digestRunLaunchValue(budgetPolicies), + executionTreeDigest: digestRunLaunchValue(executionTree), + }, }); + if (decision.outcome === 'queued' && decision.queueEntry) { + return { + schemaVersion: WORKFLOW_ADMISSION_SCHEMA_VERSION, + state: 'waiting', + workspaceId, + rootTaskId, + admissionTaskId, + attemptId, + queueEntryId: decision.queueEntry.id, + decision, + executionTree, + }; + } if (decision.outcome !== 'admitted' || !decision.reservation) { throw this.admissionConflict(decision, 'Workflow root'); } @@ -468,11 +500,13 @@ export class WorkflowRunService { } return { schemaVersion: WORKFLOW_ADMISSION_SCHEMA_VERSION, + state: 'active', workspaceId, rootTaskId, admissionTaskId, attemptId, reservationId: reservation.id, + decision, executionTree, }; } @@ -519,12 +553,13 @@ export class WorkflowRunService { step: WorkflowStep, preparation: WorkflowAgentStepPreparation ): Promise> { - if (!run.admission) { + if (!run.admission?.reservationId || run.admission.state === 'waiting') { throw new ConflictError('Workflow root admission is missing before step launch.', { runId: run.id, stepId: step.id, }); } + const rootReservationId = run.admission.reservationId; const stepRun = run.steps.find((candidate) => candidate.stepId === step.id); if (!stepRun) throw new Error(`Workflow run is missing step state for ${step.id}`); const sequence = (stepRun.admission?.sequence ?? 0) + 1; @@ -548,7 +583,7 @@ export class WorkflowRunService { const hostId = preparation.hostRouting.selectedHostId ?? (preparation.runtimeProvider === 'openclaw' ? 'openclaw-gateway' : 'local-process'); - const decision = await this.admission.admit({ + const admissionInput = { taskId: admissionTaskId, rootTaskId: run.admission.rootTaskId, workspaceId: run.admission.workspaceId, @@ -562,24 +597,53 @@ export class WorkflowRunService { : 'workflow', workflowRunId: run.id, workflowStepId: step.id, - rootReservationId: run.admission.reservationId, - idempotencyKey: `workflow-step:${run.id}:${step.id}:${sequence}:${randomUUID()}`, + rootReservationId, + idempotencyKey: `workflow-step:${run.id}:${step.id}:${sequence}`, executionTree, - budgetPolicies: (await this.admission.get(run.admission.reservationId)).request - .budgetPolicies, + budgetPolicies: (await this.admission.get(rootReservationId)).request.budgetPolicies, budgetRequest: { fanOut: Math.max(1, step.parallel?.steps.length ?? 1), retries: stepRun.runRetry ? 1 : 0, }, + } as const; + const decision = await this.admission.admitOrQueue(admissionInput, { + attemptId, + target: { + kind: 'workflow-step', + workflowId: run.workflowId, + workflowVersion: run.workflowVersion, + workflowRunId: run.id, + workflowRunRevision: (run.revision ?? 0) + 1, + workflowStepId: step.id, + workflowStepSequence: sequence, + recoverySequence: stepRun.runRetry?.sequence ?? 0, + parentNodeId: parentExecutionTree.nodeId, + edge: executionTree.edge, + provider: preparation.runtimeProvider, + hostId, + providerRuntimeManifestDigest: preparation.runtimeManifest.digest, + requiredRuntimeCapabilitiesDigest: digestRunLaunchValue( + preparation.requiredRuntimeCapabilities + ), + phaseEvidenceDigest: preparation.phaseAuthority.evidence.digest, + phaseLaunchDigest: preparation.phaseLaunchDigest, + }, }); const binding: NonNullable = { schemaVersion: WORKFLOW_ADMISSION_SCHEMA_VERSION, + state: decision.outcome === 'queued' ? 'waiting' : undefined, sequence, admissionTaskId, attemptId, decision, executionTree, }; + if (decision.outcome === 'queued' && decision.queueEntry) { + return { + ...binding, + queueEntryId: decision.queueEntry.id, + }; + } if (decision.outcome !== 'admitted' || !decision.reservation) { throw new WorkflowStepAdmissionError(binding); } @@ -599,6 +663,7 @@ export class WorkflowRunService { }); return { ...binding, + state: 'active', reservationId: reservation.id, }; } catch (error) { @@ -631,6 +696,657 @@ export class WorkflowRunService { await this.admission.release(run.admission.reservationId, reason, idempotencyKey); } + async dispatchQueuedAdmission(claim: AdmissionQueueClaim): Promise { + const target = claim.entry.target; + if (!target || target.kind === 'direct') { + throw new ConflictError('Queue claim is not a workflow launch.', { + code: 'ADMISSION_QUEUE_TARGET_MISMATCH', + queueId: claim.entry.id, + }); + } + + try { + if (target.kind === 'workflow-root') { + await this.dispatchQueuedWorkflowRoot(claim); + return; + } + await this.dispatchQueuedWorkflowStep(claim); + } catch (error) { + const current = await this.admission.getQueueEntry(claim.entry.id); + if (current.state === 'dispatched') { + log.warn( + { err: error, queueId: current.id }, + 'Workflow queue dispatch failed after durable ownership transfer' + ); + return; + } + await this.rollbackWorkflowQueueClaim(claim).catch((rollbackError) => { + log.error( + { err: rollbackError, queueId: claim.entry.id }, + 'Workflow queue claim rollback failed' + ); + }); + await this.admission + .release( + claim.reservation.id, + 'start-failed', + `workflow-queue-dispatch-failed:${claim.entry.id}` + ) + .catch(() => {}); + if ( + error instanceof ConflictError || + error instanceof NotFoundError || + error instanceof ValidationError + ) { + const terminalEntry = await this.admission.terminateQueueEntry( + claim.entry.id, + 'WORKFLOW_QUEUE_AUTHORITY_DRIFT', + 'Workflow queue authority changed before dispatch.' + ); + await this.terminalizeWorkflowQueueTarget(claim, terminalEntry); + return; + } + await this.admission.requeueQueueEntry( + claim.entry.id, + 'WORKFLOW_QUEUE_TRANSIENT_FAILURE', + 'Workflow queue dispatch failed before ownership became durable.' + ); + } + } + + private async terminalizeWorkflowQueueTarget( + claim: AdmissionQueueClaim, + terminalEntry: AdmissionQueueEntry + ): Promise { + const target = claim.entry.target; + if (!target || target.kind === 'direct') return; + const run = await this.getRun(target.workflowRunId); + if (!run) return; + const reason = terminalEntry.terminal?.reason ?? 'Workflow queue authority changed.'; + if ( + target.kind === 'workflow-root' && + run.admission?.queueEntryId === claim.entry.id && + ['waiting', 'dispatching'].includes(run.admission.state ?? '') + ) { + run.admission = { + ...run.admission, + state: 'terminal', + reservationId: undefined, + ...(run.admission.decision + ? { + decision: { + ...run.admission.decision, + queueEntry: terminalEntry, + }, + } + : {}), + }; + run.status = 'failed'; + run.error = reason; + run.completedAt = new Date().toISOString(); + await this.saveRun(run); + broadcastWorkflowStatus(run); + return; + } + if (target.kind !== 'workflow-step') return; + const stepRun = run.steps.find( + (candidate) => candidate.admission?.queueEntryId === claim.entry.id + ); + const stepAdmission = stepRun?.admission; + if ( + !stepRun || + !stepAdmission || + !['waiting', 'dispatching'].includes(stepAdmission.state ?? '') + ) { + return; + } + stepRun.admission = { + ...stepAdmission, + state: 'terminal', + reservationId: undefined, + decision: { + ...stepAdmission.decision, + queueEntry: terminalEntry, + }, + }; + stepRun.status = 'failed'; + stepRun.error = reason; + stepRun.completedAt = new Date().toISOString(); + run.status = 'failed'; + run.error = reason; + run.completedAt = stepRun.completedAt; + await this.saveRun(run); + await this.releaseRootAdmission( + run, + 'failed', + `workflow-step-queue-terminal:${run.id}:${stepRun.stepId}:${stepAdmission.sequence}` + ).catch((releaseError) => { + log.error( + { err: releaseError, runId: run.id, stepId: stepRun.stepId }, + 'Failed to release workflow root after terminal queue drift' + ); + }); + broadcastWorkflowStatus(run); + } + + private async dispatchQueuedWorkflowRoot(claim: AdmissionQueueClaim): Promise { + const target = claim.entry.target; + if (target?.kind !== 'workflow-root') { + throw new ConflictError('Queue claim is not a workflow root.', { + code: 'ADMISSION_QUEUE_TARGET_MISMATCH', + queueId: claim.entry.id, + }); + } + const run = await this.getRun(target.workflowRunId); + if (!run) throw new NotFoundError(`Run ${target.workflowRunId} not found`); + const workflow = await this.workflowService.loadWorkflow(run.workflowId); + if (!workflow) throw new NotFoundError(`Workflow ${run.workflowId} not found`); + const driftFields = [ + run.workflowId !== target.workflowId ? 'workflowId' : undefined, + run.workflowVersion !== target.workflowVersion ? 'workflowVersion' : undefined, + workflow.version !== target.workflowVersion ? 'currentWorkflowVersion' : undefined, + run.revision !== target.workflowRunRevision ? 'workflowRunRevision' : undefined, + run.taskId !== target.associatedTaskId ? 'associatedTaskId' : undefined, + run.status !== 'pending' ? 'runStatus' : undefined, + run.admission?.state !== 'waiting' ? 'admissionState' : undefined, + run.admission?.queueEntryId !== claim.entry.id ? 'queueEntryId' : undefined, + run.admission?.attemptId !== claim.entry.attemptId ? 'attemptId' : undefined, + digestRunLaunchValue(run.context) !== target.initialContextDigest + ? 'initialContextDigest' + : undefined, + digestRunLaunchValue(claim.entry.request.budgetPolicies ?? []) !== target.budgetPolicyDigest + ? 'budgetPolicyDigest' + : undefined, + digestRunLaunchValue(run.executionTree) !== target.executionTreeDigest + ? 'executionTreeDigest' + : undefined, + ].filter((field): field is string => Boolean(field)); + if (driftFields.length > 0) { + throw new ConflictError('Queued workflow root changed before dispatch.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: claim.entry.id, + driftFields, + }); + } + + const reservation = await this.admission.bindQueuedAttempt( + claim.entry.id, + claim.reservation.id, + claim.entry.attemptId + ); + await this.admission.recordBudgetUsage(reservation.id, { + schemaVersion: 'execution-tree-budget-event/v1', + id: `launch_${claim.entry.attemptId}`, + mode: 'delta', + usage: { ...ZERO_AGENT_BUDGET_USAGE, fanOut: 1 }, + source: 'workflow-root-launch', + occurredAt: run.startedAt, + }); + const currentAdmission = run.admission; + if (!currentAdmission) { + throw new ConflictError('Queued workflow root admission binding is missing.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: claim.entry.id, + }); + } + const dispatchingAdmission: NonNullable = { + ...currentAdmission, + state: 'dispatching', + reservationId: reservation.id, + }; + run.admission = dispatchingAdmission; + await this.saveRun(run); + await this.admission.markQueueDispatched(claim.entry.id, claim.entry.attemptId); + run.admission = { ...dispatchingAdmission, state: 'active' }; + run.status = 'running'; + run.error = undefined; + await this.saveRun(run); + + void this.executeRun(run, workflow).catch((error) => { + log.error({ err: error, runId: run.id }, 'Queued workflow root execution failed'); + }); + } + + private async dispatchQueuedWorkflowStep(claim: AdmissionQueueClaim): Promise { + const target = claim.entry.target; + if (target?.kind !== 'workflow-step') { + throw new ConflictError('Queue claim is not a workflow step.', { + code: 'ADMISSION_QUEUE_TARGET_MISMATCH', + queueId: claim.entry.id, + }); + } + const run = await this.getRun(target.workflowRunId); + if (!run) throw new NotFoundError(`Run ${target.workflowRunId} not found`); + const workflow = await this.workflowService.loadWorkflow(run.workflowId); + if (!workflow) throw new NotFoundError(`Workflow ${run.workflowId} not found`); + const step = workflow.steps.find((candidate) => candidate.id === target.workflowStepId); + if (!step) throw new NotFoundError(`Workflow step ${target.workflowStepId} not found`); + const stepRun = run.steps.find((candidate) => candidate.stepId === target.workflowStepId); + if (!stepRun) throw new NotFoundError(`Run step ${target.workflowStepId} not found`); + const driftFields = [ + run.workflowId !== target.workflowId ? 'workflowId' : undefined, + run.workflowVersion !== target.workflowVersion ? 'workflowVersion' : undefined, + workflow.version !== target.workflowVersion ? 'currentWorkflowVersion' : undefined, + run.revision !== target.workflowRunRevision ? 'workflowRunRevision' : undefined, + run.status !== 'pending' ? 'runStatus' : undefined, + run.currentStep !== target.workflowStepId ? 'currentStep' : undefined, + !run.admission?.reservationId || run.admission.state === 'waiting' + ? 'rootAdmission' + : undefined, + claim.entry.request.rootReservationId !== run.admission?.reservationId + ? 'rootReservationId' + : undefined, + stepRun.status !== 'pending' ? 'stepStatus' : undefined, + stepRun.admission?.state !== 'waiting' ? 'stepAdmissionState' : undefined, + stepRun.admission?.queueEntryId !== claim.entry.id ? 'queueEntryId' : undefined, + stepRun.admission?.sequence !== target.workflowStepSequence + ? 'workflowStepSequence' + : undefined, + (stepRun.runRetry?.sequence ?? 0) !== target.recoverySequence + ? 'recoverySequence' + : undefined, + stepRun.executionTree?.parentNodeId !== target.parentNodeId ? 'parentNodeId' : undefined, + stepRun.executionTree?.edge !== target.edge ? 'executionTreeEdge' : undefined, + ].filter((field): field is string => Boolean(field)); + if (driftFields.length > 0) { + throw new ConflictError('Queued workflow step changed before dispatch.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: claim.entry.id, + driftFields, + }); + } + + const preparation = await this.stepExecutor.prepareStep(step, run); + if (preparation.kind !== 'agent') { + throw new ConflictError('Queued workflow step is no longer provider-backed.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: claim.entry.id, + }); + } + this.assertQueuedWorkflowStepPreparation(run, stepRun, preparation); + const reservation = await this.admission.bindQueuedAttempt( + claim.entry.id, + claim.reservation.id, + claim.entry.attemptId + ); + await this.admission.recordBudgetUsage(reservation.id, { + schemaVersion: 'execution-tree-budget-event/v1', + id: `launch_${claim.entry.attemptId}`, + mode: 'delta', + usage: { + ...ZERO_AGENT_BUDGET_USAGE, + fanOut: Math.max(1, step.parallel?.steps.length ?? 1), + retries: stepRun.runRetry ? 1 : 0, + }, + source: 'workflow-step-launch', + occurredAt: run.startedAt, + }); + this.stepExecutor.applyPreparation(run, preparation); + const currentStepAdmission = stepRun.admission; + if (!currentStepAdmission) { + throw new ConflictError('Queued workflow step admission binding is missing.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: claim.entry.id, + }); + } + stepRun.admission = { + ...currentStepAdmission, + state: 'dispatching', + reservationId: reservation.id, + }; + stepRun.error = undefined; + run.status = 'pending'; + run.error = undefined; + await this.saveRun(run); + await this.admission.markQueueDispatched(claim.entry.id, claim.entry.attemptId); + + void this.executeRun(run, workflow).catch((error) => { + log.error( + { err: error, runId: run.id, stepId: step.id }, + 'Queued workflow step execution failed' + ); + }); + } + + private assertQueuedWorkflowStepPreparation( + run: WorkflowRun, + stepRun: StepRun, + preparation: WorkflowAgentStepPreparation + ): void { + const target = stepRun.admission?.decision.queueEntry?.target; + if (target?.kind !== 'workflow-step') { + throw new ConflictError('Workflow step queue target is missing.', { + code: 'ADMISSION_QUEUE_DRIFT', + runId: run.id, + stepId: stepRun.stepId, + }); + } + const hostId = + preparation.hostRouting.selectedHostId ?? + (preparation.runtimeProvider === 'openclaw' ? 'openclaw-gateway' : 'local-process'); + const driftFields = [ + preparation.runtimeProvider !== target.provider ? 'provider' : undefined, + hostId !== target.hostId ? 'hostId' : undefined, + preparation.runtimeManifest.digest !== target.providerRuntimeManifestDigest + ? 'providerRuntimeManifestDigest' + : undefined, + digestRunLaunchValue(preparation.requiredRuntimeCapabilities) !== + target.requiredRuntimeCapabilitiesDigest + ? 'requiredRuntimeCapabilitiesDigest' + : undefined, + preparation.phaseAuthority.evidence.digest !== target.phaseEvidenceDigest + ? 'phaseEvidenceDigest' + : undefined, + preparation.phaseLaunchDigest !== target.phaseLaunchDigest ? 'phaseLaunchDigest' : undefined, + ].filter((field): field is string => Boolean(field)); + if (driftFields.length > 0) { + throw new ConflictError('Queued workflow step launch authority changed.', { + code: 'ADMISSION_QUEUE_DRIFT', + queueId: stepRun.admission?.queueEntryId, + driftFields, + }); + } + } + + private async rollbackWorkflowQueueClaim(claim: AdmissionQueueClaim): Promise { + const target = claim.entry.target; + if (!target || target.kind === 'direct') return; + const run = await this.getRun(target.workflowRunId); + if (!run) return; + if ( + run.admission?.queueEntryId === claim.entry.id && + run.admission.state === 'dispatching' && + run.admission.reservationId === claim.reservation.id + ) { + run.admission = { + ...run.admission, + state: 'waiting', + reservationId: undefined, + }; + run.status = 'pending'; + run.error = 'Workflow launch is waiting for admission capacity.'; + await this.saveRun(run); + return; + } + const stepRun = run.steps.find( + (candidate) => candidate.admission?.queueEntryId === claim.entry.id + ); + if ( + stepRun?.admission?.state === 'dispatching' && + stepRun.admission.reservationId === claim.reservation.id + ) { + stepRun.admission = { + ...stepRun.admission, + state: 'waiting', + reservationId: undefined, + }; + stepRun.status = 'pending'; + stepRun.error = 'Workflow step is waiting for admission capacity.'; + run.status = 'pending'; + run.error = stepRun.error; + await this.saveRun(run); + } + } + + private async reconcileWorkflowRootQueue(run: WorkflowRun): Promise { + const binding = run.admission; + if (!binding || !['waiting', 'dispatching'].includes(binding.state ?? '')) { + return false; + } + if (!binding.queueEntryId) { + throw new ConflictError('Workflow queue binding is missing its queue entry.', { + runId: run.id, + admissionState: binding.state, + }); + } + const entry = await this.admission.getQueueEntry(binding.queueEntryId); + if (binding.state === 'waiting') { + if (entry.state === 'terminal') { + run.admission = { + ...binding, + state: 'terminal', + decision: binding.decision + ? { ...binding.decision, queueEntry: entry } + : binding.decision, + }; + run.status = 'failed'; + run.error = entry.terminal?.reason ?? 'Workflow admission queue terminated.'; + run.completedAt = new Date().toISOString(); + await this.saveRun(run); + broadcastWorkflowStatus(run); + } else { + this.scheduleAdmissionQueueDrain(); + } + return true; + } + if ( + entry.state === 'dispatched' && + entry.dispatchedAttemptId === binding.attemptId && + entry.reservationId === binding.reservationId + ) { + const recovered = await this.admission.recoverVerifiedRun({ + workspaceId: binding.workspaceId, + taskId: binding.admissionTaskId, + attemptId: binding.attemptId, + }); + if (!recovered || recovered.id !== binding.reservationId) { + throw new ConflictError('Dispatched workflow root admission could not be recovered.', { + runId: run.id, + queueId: entry.id, + expectedReservationId: binding.reservationId, + recoveredReservationId: recovered?.id, + }); + } + const workflow = await this.workflowService.loadWorkflow(run.workflowId); + if (!workflow || workflow.version !== run.workflowVersion) { + throw new ConflictError('Dispatched workflow definition is unavailable for recovery.', { + runId: run.id, + workflowId: run.workflowId, + workflowVersion: run.workflowVersion, + currentWorkflowVersion: workflow?.version, + }); + } + run.admission = { ...binding, state: 'active' }; + run.status = 'running'; + run.error = undefined; + await this.saveRun(run); + void this.executeRun(run, workflow).catch((error) => { + log.error({ err: error, runId: run.id }, 'Recovered workflow root execution failed'); + }); + return true; + } + if (entry.state === 'terminal') { + run.admission = { + ...binding, + state: 'terminal', + reservationId: undefined, + decision: binding.decision ? { ...binding.decision, queueEntry: entry } : binding.decision, + }; + run.status = 'failed'; + run.error = entry.terminal?.reason ?? 'Workflow admission queue terminated.'; + run.completedAt = new Date().toISOString(); + await this.saveRun(run); + if (binding.reservationId) { + await this.admission + .release( + binding.reservationId, + 'start-failed', + `workflow-root-queue-terminal-restart:${run.id}` + ) + .catch(() => {}); + } + broadcastWorkflowStatus(run); + return true; + } + run.admission = { + ...binding, + state: 'waiting', + reservationId: undefined, + }; + run.status = 'pending'; + run.error = 'Workflow root is waiting for admission capacity.'; + await this.saveRun(run); + if (binding.reservationId) { + await this.admission.release( + binding.reservationId, + 'start-failed', + `workflow-root-queue-restart:${run.id}` + ); + } + if (entry.state === 'leased') { + await this.admission.requeueQueueEntry( + entry.id, + 'WORKFLOW_QUEUE_RESTART', + 'Server restarted before workflow ownership became durable.' + ); + } + this.scheduleAdmissionQueueDrain(); + broadcastWorkflowStatus(run); + return true; + } + + private async reconcileWorkflowStepQueue(run: WorkflowRun, stepRun: StepRun): Promise { + const binding = stepRun.admission; + if (!binding || !['waiting', 'dispatching'].includes(binding.state ?? '')) { + return false; + } + if (!binding.queueEntryId) { + throw new ConflictError('Workflow step queue binding is missing its queue entry.', { + runId: run.id, + stepId: stepRun.stepId, + admissionState: binding.state, + }); + } + const entry = await this.admission.getQueueEntry(binding.queueEntryId); + if (binding.state === 'waiting') { + if (entry.state === 'terminal') { + stepRun.admission = { + ...binding, + state: 'terminal', + decision: { ...binding.decision, queueEntry: entry }, + }; + stepRun.status = 'failed'; + stepRun.error = entry.terminal?.reason ?? 'Workflow step admission queue terminated.'; + stepRun.completedAt = new Date().toISOString(); + run.status = 'failed'; + run.error = stepRun.error; + run.completedAt = stepRun.completedAt; + await this.saveRun(run); + await this.releaseRootAdmission( + run, + 'failed', + `workflow-step-queue-terminal-restart:${run.id}:${stepRun.stepId}:${binding.sequence}` + ).catch(() => {}); + broadcastWorkflowStatus(run); + } else { + this.scheduleAdmissionQueueDrain(); + } + return true; + } + if ( + entry.state === 'dispatched' && + entry.dispatchedAttemptId === binding.attemptId && + entry.reservationId === binding.reservationId + ) { + if (!binding.reservationId) { + throw new ConflictError('Dispatched workflow step is missing its reservation.', { + runId: run.id, + stepId: stepRun.stepId, + queueId: entry.id, + }); + } + const persisted = await this.admission.get(binding.reservationId); + const recovered = await this.admission.recoverVerifiedRun({ + workspaceId: persisted.request.workspaceId, + taskId: binding.admissionTaskId, + attemptId: binding.attemptId, + }); + if (!recovered || recovered.id !== binding.reservationId) { + throw new ConflictError('Dispatched workflow step admission could not be recovered.', { + runId: run.id, + stepId: stepRun.stepId, + queueId: entry.id, + expectedReservationId: binding.reservationId, + recoveredReservationId: recovered?.id, + }); + } + const workflow = await this.workflowService.loadWorkflow(run.workflowId); + if (!workflow || workflow.version !== run.workflowVersion) { + throw new ConflictError('Dispatched workflow definition is unavailable for recovery.', { + runId: run.id, + workflowId: run.workflowId, + workflowVersion: run.workflowVersion, + currentWorkflowVersion: workflow?.version, + }); + } + void this.executeRun(run, workflow).catch((error) => { + log.error( + { err: error, runId: run.id, stepId: stepRun.stepId }, + 'Recovered workflow step execution failed' + ); + }); + return true; + } + if (entry.state === 'terminal') { + stepRun.admission = { + ...binding, + state: 'terminal', + reservationId: undefined, + decision: { ...binding.decision, queueEntry: entry }, + }; + stepRun.status = 'failed'; + stepRun.error = entry.terminal?.reason ?? 'Workflow step admission queue terminated.'; + stepRun.completedAt = new Date().toISOString(); + run.status = 'failed'; + run.error = stepRun.error; + run.completedAt = stepRun.completedAt; + await this.saveRun(run); + if (binding.reservationId) { + await this.admission + .release( + binding.reservationId, + 'start-failed', + `workflow-step-queue-terminal-restart:${run.id}:${stepRun.stepId}:${binding.sequence}` + ) + .catch(() => {}); + } + await this.releaseRootAdmission( + run, + 'failed', + `workflow-step-queue-terminal-root:${run.id}:${stepRun.stepId}:${binding.sequence}` + ).catch(() => {}); + broadcastWorkflowStatus(run); + return true; + } + stepRun.admission = { + ...binding, + state: 'waiting', + reservationId: undefined, + }; + stepRun.status = 'pending'; + stepRun.error = 'Workflow step is waiting for admission capacity.'; + run.status = 'pending'; + run.error = stepRun.error; + await this.saveRun(run); + if (binding.reservationId) { + await this.admission.release( + binding.reservationId, + 'start-failed', + `workflow-step-queue-restart:${run.id}:${stepRun.stepId}:${binding.sequence}` + ); + } + if (entry.state === 'leased') { + await this.admission.requeueQueueEntry( + entry.id, + 'WORKFLOW_QUEUE_RESTART', + 'Server restarted before workflow step ownership became durable.' + ); + } + this.scheduleAdmissionQueueDrain(); + broadcastWorkflowStatus(run); + return true; + } + /** * Start a new workflow run */ @@ -749,9 +1465,18 @@ export class WorkflowRunService { try { run.admission = await this.admitWorkflowRoot(run, task, budgetSources); - run.status = 'running'; + const waitingForAdmission = run.admission.state === 'waiting'; + run.status = waitingForAdmission ? 'pending' : 'running'; + run.error = waitingForAdmission + ? 'Workflow root is waiting for admission capacity.' + : undefined; await this.saveRun(run); await this.snapshotWorkflow(run.id, workflow); + if (waitingForAdmission) { + this.scheduleAdmissionQueueDrain(); + broadcastWorkflowStatus(run); + return run; + } } catch (error) { if (run.admission) { if (run.revision) { @@ -781,6 +1506,20 @@ export class WorkflowRunService { return run; } + private scheduleAdmissionQueueDrain(): void { + if (this.requestAdmissionQueueDrain) { + this.requestAdmissionQueueDrain(); + return; + } + queueMicrotask(() => { + void import('./clawdbot-agent-service.js') + .then(({ clawdbotAgentService }) => clawdbotAgentService.reconcileQueuedLaunches()) + .catch((error) => { + log.error({ err: error }, 'Workflow admission queue drain failed'); + }); + }); + } + /** * Execute the workflow run (iterates through steps with retry logic) */ @@ -826,7 +1565,25 @@ export class WorkflowRunService { // or mutating the executable step attempt. const preparation = await this.stepExecutor.prepareStep(step, run); if (preparation.kind === 'agent') { - stepRun.admission = await this.admitWorkflowStep(run, step, preparation); + if (stepRun.admission?.state === 'dispatching') { + this.assertQueuedWorkflowStepPreparation(run, stepRun, preparation); + stepRun.admission = { ...stepRun.admission, state: 'active' }; + } else { + stepRun.admission = await this.admitWorkflowStep(run, step, preparation); + } + if (stepRun.admission.state === 'waiting') { + run.status = 'pending'; + run.currentStep = step.id; + run.error = 'Workflow step is waiting for admission capacity.'; + stepRun.status = 'pending'; + stepRun.error = run.error; + stepRun.completedAt = undefined; + this.syncPipelineSummary(run, workflow); + await this.saveRun(run); + broadcastWorkflowStatus(run); + this.scheduleAdmissionQueueDrain(); + return; + } } run.status = 'running'; run.currentStep = step.id; @@ -1296,6 +2053,9 @@ export class WorkflowRunService { let scheduledCount = 0; for (const run of runs) { try { + if (await this.reconcileWorkflowRootQueue(run)) { + continue; + } await this.ensureWorkflowRootAdmission(run); } catch (error) { run.status = 'blocked'; @@ -1308,6 +2068,22 @@ export class WorkflowRunService { continue; } const stepRun = run.steps.find((step) => step.stepId === run.currentStep); + if (stepRun) { + try { + if (await this.reconcileWorkflowStepQueue(run, stepRun)) { + continue; + } + } catch (error) { + run.status = 'blocked'; + run.error = + error instanceof Error + ? `Workflow step admission recovery failed: ${error.message}` + : 'Workflow step admission recovery failed.'; + await this.saveRun(run); + broadcastWorkflowStatus(run); + continue; + } + } const recovery = stepRun?.runRetry; if (stepRun?.status === 'running') { await this.releaseStepAdmission( @@ -2066,6 +2842,7 @@ export interface WorkflowRunServiceOptions { runRecoveryPolicy?: RunRecoveryPolicyService; stepExecutor?: WorkflowStepExecutor; admission?: AdmissionControlService; + requestAdmissionQueueDrain?: () => void; } function workflowExecutionTreePolicy( diff --git a/server/src/storage/admission-queue-identity.ts b/server/src/storage/admission-queue-identity.ts new file mode 100644 index 00000000..ee464996 --- /dev/null +++ b/server/src/storage/admission-queue-identity.ts @@ -0,0 +1,14 @@ +import type { AdmissionQueueDraft, AdmissionQueueEntry } from '@veritas-kanban/shared'; + +type QueueIdentity = Pick; + +export function sameAdmissionQueueTarget( + existing: QueueIdentity, + requested: Pick +): boolean { + const existingTarget = + existing.target ?? (existing.agent ? { kind: 'direct', agent: existing.agent } : undefined); + const requestedTarget = + requested.target ?? (requested.agent ? { kind: 'direct', agent: requested.agent } : undefined); + return JSON.stringify(existingTarget) === JSON.stringify(requestedTarget); +} diff --git a/server/src/storage/admission-reservation-repository.ts b/server/src/storage/admission-reservation-repository.ts index 3483dc91..d226b1ad 100644 --- a/server/src/storage/admission-reservation-repository.ts +++ b/server/src/storage/admission-reservation-repository.ts @@ -27,8 +27,10 @@ import { withFileLock } from '../services/file-lock.js'; import { getRuntimeDir } from '../utils/paths.js'; import { ensureWithinBase } from '../utils/sanitize.js'; import { findLimitingAdmissionPolicies } from './admission-capacity.js'; +import { sameAdmissionQueueTarget } from './admission-queue-identity.js'; import { findLimitingExecutionTreeBudgetPolicies, + reactivateExecutionTreeBudget, releaseExecutionTreeBudget, } from './execution-tree-budget.js'; import type { AdmissionReservationRepository } from './interfaces.js'; @@ -97,7 +99,7 @@ export class FileAdmissionReservationRepository implements AdmissionReservationR ); if (existing) { if (expired.length > 0) await this.replaceQueueEntries([...queue.values()]); - if (existing.agent !== input.queue.agent) { + if (!sameAdmissionQueueTarget(existing, input.queue)) { return { ...claimed, queueConflict: true }; } return { ...claimed, queueEntry: existing }; @@ -163,7 +165,12 @@ export class FileAdmissionReservationRepository implements AdmissionReservationR } const claimed = await this.claimMaterialized( requested, - { record: requested, now: input.now, reclaimExpired: true }, + { + record: requested, + now: input.now, + reclaimExpired: true, + reclaimReleased: true, + }, reservations, snapshots ); @@ -207,13 +214,20 @@ export class FileAdmissionReservationRepository implements AdmissionReservationR if (existing.request.idempotencyKey !== requested.request.idempotencyKey) { throw new Error(`Admission reservation ${requested.id} has conflicting identity.`); } - if (existing.state !== 'expired' || !input.reclaimExpired) { + const reclaimable = + (existing.state === 'expired' && input.reclaimExpired) || + (existing.state === 'released' && input.reclaimReleased); + if (!reclaimable) { return { record: existing, created: false, limitingPolicies: [] }; } const reclaimed = AdmissionReservationSchema.parse({ ...requested, revision: existing.revision + 1, createdAt: existing.createdAt, + executionBudget: reactivateExecutionTreeBudget( + existing.executionBudget, + requested.executionBudget + ), }); const limitingPolicies = findLimitingAdmissionPolicies( [...materialized.values()].filter((record) => record.id !== existing.id), diff --git a/server/src/storage/execution-tree-budget.ts b/server/src/storage/execution-tree-budget.ts index 79d2c55e..c3521d74 100644 --- a/server/src/storage/execution-tree-budget.ts +++ b/server/src/storage/execution-tree-budget.ts @@ -95,6 +95,21 @@ export function releaseExecutionTreeBudget( }; } +export function reactivateExecutionTreeBudget( + existing: ExecutionTreeBudgetState | undefined, + requested: ExecutionTreeBudgetState | undefined +): ExecutionTreeBudgetState | undefined { + if (!existing || !requested) return requested; + const remaining = subtractBudgetUsage(requested.requested, existing.committed); + return { + requested: { ...requested.requested }, + remaining, + committed: { ...existing.committed }, + releasedUnused: subtractBudgetUsage(existing.releasedUnused, remaining), + events: [...existing.events], + }; +} + function sameExecutionTree(record: AdmissionReservation, candidate: AdmissionReservation): boolean { return ( record.request.executionTree?.rootObjectiveId === diff --git a/server/src/storage/sqlite/admission-reservation-repository.ts b/server/src/storage/sqlite/admission-reservation-repository.ts index 14677705..25fe8f06 100644 --- a/server/src/storage/sqlite/admission-reservation-repository.ts +++ b/server/src/storage/sqlite/admission-reservation-repository.ts @@ -20,8 +20,10 @@ import { AdmissionReservationSchema, } from '../../schemas/admission-control-schemas.js'; import { findLimitingAdmissionPolicies } from '../admission-capacity.js'; +import { sameAdmissionQueueTarget } from '../admission-queue-identity.js'; import { findLimitingExecutionTreeBudgetPolicies, + reactivateExecutionTreeBudget, releaseExecutionTreeBudget, } from '../execution-tree-budget.js'; import type { AdmissionReservationRepository } from '../interfaces.js'; @@ -84,7 +86,7 @@ export class SqliteAdmissionReservationRepository implements AdmissionReservatio if (existingRow) { const existing = AdmissionQueueEntrySchema.parse(JSON.parse(existingRow.queue_json)); connection.exec('COMMIT'); - if (existing.agent !== input.queue.agent) { + if (!sameAdmissionQueueTarget(existing, input.queue)) { return { ...claimed, queueConflict: true }; } return { @@ -168,6 +170,7 @@ export class SqliteAdmissionReservationRepository implements AdmissionReservatio record: requested, now: input.now, reclaimExpired: true, + reclaimReleased: true, }); if (!claimed.record) { connection.exec('COMMIT'); @@ -388,13 +391,20 @@ export class SqliteAdmissionReservationRepository implements AdmissionReservatio if (existing.request.idempotencyKey !== requested.request.idempotencyKey) { throw new Error(`Admission reservation ${requested.id} has conflicting identity.`); } - if (existing.state !== 'expired' || !input.reclaimExpired) { + const reclaimable = + (existing.state === 'expired' && input.reclaimExpired) || + (existing.state === 'released' && input.reclaimReleased); + if (!reclaimable) { return { record: existing, created: false, limitingPolicies: [] }; } const reclaimed = AdmissionReservationSchema.parse({ ...requested, revision: existing.revision + 1, createdAt: existing.createdAt, + executionBudget: reactivateExecutionTreeBudget( + existing.executionBudget, + requested.executionBudget + ), }); const limitingPolicies = findLimitingAdmissionPolicies( this.activeReservations().filter((record) => record.id !== existing.id), diff --git a/shared/src/types/admission-control.types.ts b/shared/src/types/admission-control.types.ts index fb33ac7c..b100a35f 100644 --- a/shared/src/types/admission-control.types.ts +++ b/shared/src/types/admission-control.types.ts @@ -133,13 +133,50 @@ export interface AdmissionQueueTerminalEvidence { recordedAt: string; } +export type AdmissionQueueTarget = + | { + kind: 'direct'; + agent: AgentType; + } + | { + kind: 'workflow-root'; + workflowId: string; + workflowVersion: number; + workflowRunId: string; + workflowRunRevision: number; + associatedTaskId?: string; + initialContextDigest: string; + budgetPolicyDigest: string; + executionTreeDigest: string; + } + | { + kind: 'workflow-step'; + workflowId: string; + workflowVersion: number; + workflowRunId: string; + workflowRunRevision: number; + workflowStepId: string; + workflowStepSequence: number; + recoverySequence: number; + parentNodeId: string; + edge: ExecutionTreeIdentity['edge']; + provider: ExecutableAgentProvider; + hostId: string; + providerRuntimeManifestDigest: string; + requiredRuntimeCapabilitiesDigest: string; + phaseEvidenceDigest: string; + phaseLaunchDigest: string; + }; + export interface AdmissionQueueEntry { schemaVersion: typeof ADMISSION_QUEUE_ENTRY_SCHEMA_VERSION; id: string; revision: number; state: AdmissionQueueState; enqueueSequence: number; - agent: AgentType; + /** Legacy direct-launch discriminator. New entries also persist target. */ + agent?: AgentType; + target?: AdmissionQueueTarget; attemptId: string; request: AdmissionRequest; policies: AdmissionLimitPolicy[]; @@ -180,7 +217,8 @@ export interface AdmissionQueueListQuery { export interface AdmissionQueueDraft { id: string; - agent: AgentType; + agent?: AgentType; + target?: AdmissionQueueTarget; attemptId: string; request: AdmissionRequest; policies: AdmissionLimitPolicy[]; @@ -224,6 +262,7 @@ export interface AdmissionReservationClaimInput { record: AdmissionReservation; now: string; reclaimExpired?: boolean; + reclaimReleased?: boolean; } export interface AdmissionReservationClaimResult { diff --git a/shared/src/types/workflow.ts b/shared/src/types/workflow.ts index f4eaf0b5..d9bac625 100644 --- a/shared/src/types/workflow.ts +++ b/shared/src/types/workflow.ts @@ -244,23 +244,29 @@ export interface ParallelSubStep { export type WorkflowRunStatus = 'pending' | 'running' | 'blocked' | 'completed' | 'failed'; export type StepRunStatus = 'pending' | 'running' | 'completed' | 'failed' | 'skipped'; export const WORKFLOW_ADMISSION_SCHEMA_VERSION = 'workflow-admission/v1' as const; +export type WorkflowAdmissionState = 'waiting' | 'dispatching' | 'active' | 'terminal'; export interface WorkflowRootAdmissionBinding { schemaVersion: typeof WORKFLOW_ADMISSION_SCHEMA_VERSION; + state?: WorkflowAdmissionState; workspaceId: string; rootTaskId: string; admissionTaskId: string; attemptId: string; - reservationId: string; + reservationId?: string; + queueEntryId?: string; + decision?: import('./admission-control.types.js').AdmissionDecision; executionTree: import('./execution-tree-budget.types.js').ExecutionTreeIdentity; } export interface WorkflowStepAdmissionBinding { schemaVersion: typeof WORKFLOW_ADMISSION_SCHEMA_VERSION; + state?: WorkflowAdmissionState; sequence: number; admissionTaskId: string; attemptId: string; reservationId?: string; + queueEntryId?: string; decision: import('./admission-control.types.js').AdmissionDecision; executionTree: import('./execution-tree-budget.types.js').ExecutionTreeIdentity; }