mirror of
https://github.com/BradGroux/veritas-kanban.git
synced 2026-08-28 02:44:59 +00:00
feat: queue saturated workflow launches (#1067)
This commit is contained in:
Brad Groux
GitHub
parent
b74c4db259
commit
4a2ecc5bd1
17 changed files with 1707 additions and 63 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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_<timestamp>_<nanoid>
|
||||
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<string, unknown>; // 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
|
||||
|
|
|
|||
|
|
@ -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**:
|
||||
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
|
|
|
|||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -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<ReturnType<typeof createHarness>>,
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
|
||||
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<typeof claim>) => Promise<void>;
|
||||
}
|
||||
).dispatchQueuedAdmission(claim as NonNullable<typeof claim>);
|
||||
|
||||
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 },
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<AdmissionQueueTarget, { kind: 'direct' }>;
|
||||
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<AdmissionDecision> {
|
||||
return this.admitInternal(input, queue);
|
||||
}
|
||||
|
||||
private async admitInternal(
|
||||
input: AdmissionRequestInput,
|
||||
queue?: DirectAdmissionQueueInput
|
||||
queue?: AdmissionQueueInput
|
||||
): Promise<AdmissionDecision> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<NonNullable<StepRun['admission']>> {
|
||||
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<StepRun['admission']> = {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<WorkflowRun['admission']> = {
|
||||
...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<void> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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(
|
||||
|
|
|
|||
14
server/src/storage/admission-queue-identity.ts
Normal file
14
server/src/storage/admission-queue-identity.ts
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
import type { AdmissionQueueDraft, AdmissionQueueEntry } from '@veritas-kanban/shared';
|
||||
|
||||
type QueueIdentity = Pick<AdmissionQueueEntry, 'agent' | 'target'>;
|
||||
|
||||
export function sameAdmissionQueueTarget(
|
||||
existing: QueueIdentity,
|
||||
requested: Pick<AdmissionQueueDraft, 'agent' | 'target'>
|
||||
): 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);
|
||||
}
|
||||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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 ===
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue