Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions packages/daemon/src/lib/rpc-handlers/space-task-handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
*/

import {
isRateOrUsageLimited,
isWorkflowRecoveryTransition,
resolveNodeAgents,
type CreateSpaceTaskParams,
Expand Down Expand Up @@ -469,12 +470,10 @@ export function setupSpaceTaskHandlers(
const fromActivePaused =
currentTask.status === 'in_progress' ||
currentTask.status === 'blocked' ||
currentTask.status === 'rate_limited' ||
currentTask.status === 'usage_limited';
isRateOrUsageLimited(currentTask.status);
const toStopped = updateParams.status === 'open' || updateParams.status === 'cancelled';
const toBlockedFromPaused =
updateParams.status === 'blocked' &&
(currentTask.status === 'rate_limited' || currentTask.status === 'usage_limited');
updateParams.status === 'blocked' && isRateOrUsageLimited(currentTask.status);
const shouldStopWorkflowForStatus =
!!currentTask.workflowRunId && fromActivePaused && (toStopped || toBlockedFromPaused);
// Reject bare transitions into `review`. Every task that lands in
Expand Down
4 changes: 2 additions & 2 deletions packages/daemon/src/lib/space/goals/goal-service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import type {
SpaceTask,
UpdateSpaceGoalParams,
} from '@hyperneo/shared';
import { isRateOrUsageLimited } from '@hyperneo/shared';
import type { Database as BunDatabase } from 'bun:sqlite';
import type { SpaceRepository } from '../../../storage/repositories/space-repository';
import type { SpaceTaskRepository } from '../../../storage/repositories/space-task-repository';
Expand Down Expand Up @@ -568,8 +569,7 @@ function isActiveTaskStatus(status: SpaceTask['status']): boolean {
// A task paused on a rate/usage cap is still the goal's active run — it
// auto-resumes when the cap lifts. Treating it as inactive would let the
// goal clear activeTaskId and spawn/claim a second concurrent task.
status === 'rate_limited' ||
status === 'usage_limited'
isRateOrUsageLimited(status)
);
}

Expand Down
4 changes: 2 additions & 2 deletions packages/daemon/src/lib/space/managers/space-task-manager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import type {
SpaceTaskStatus,
UpdateSpaceTaskParams,
} from '@hyperneo/shared';
import { isRateOrUsageLimited } from '@hyperneo/shared';
import type { ReactiveDatabase } from '../../../storage/reactive-database';
import { SpaceTaskRepository } from '../../../storage/repositories/space-task-repository';
import { Logger } from '../../logger';
Expand Down Expand Up @@ -749,8 +750,7 @@ export class SpaceTaskManager {
// A dependent paused on a rate/usage cap must be cancelled too —
// otherwise recoverRateLimitedTasks would later restore it to
// in_progress and resume work despite the cancelled prerequisite.
t.status === 'rate_limited' ||
t.status === 'usage_limited'
isRateOrUsageLimited(t.status)
) {
const cancelled = await this.setTaskStatus(t.id, 'cancelled', {
result: `Dependency task ${taskId} was cancelled`,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import type {
SpaceWorkflowRun,
UpdateSpaceTaskParams,
} from '@hyperneo/shared';
import { isRateOrUsageLimited } from '@hyperneo/shared';
import type { MessageRecord, ActorRef } from '../../../../../messaging/src/types';
import { canonicalAgentHandle, SpaceActorRegistryAdapter } from '../actor-registry';
import type { ExternalEventPublishedPayload } from '../../external-events/external-event-service';
Expand Down Expand Up @@ -1072,11 +1073,7 @@ export class SpaceRuntimeService {
const activeTasks = taskRepo
.listBySpace(spaceId)
.filter(
(t) =>
t.status === 'in_progress' ||
t.status === 'open' ||
t.status === 'rate_limited' ||
t.status === 'usage_limited'
(t) => t.status === 'in_progress' || t.status === 'open' || isRateOrUsageLimited(t.status)
);

await Promise.allSettled(
Expand Down
6 changes: 3 additions & 3 deletions packages/daemon/src/lib/space/runtime/space-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import type { SDKMessage } from '@hyperneo/shared/sdk';
import {
computeGateDefaults,
isChannelCyclic,
isRateOrUsageLimited,
isWorkflowRunSucceeded,
isWorkflowRunWaiting,
MAX_SPACE_CONCURRENT_TASKS,
Expand Down Expand Up @@ -7616,7 +7617,7 @@ export class SpaceRuntime {
// `recoverRateLimitedTasks()` later restores the task to `in_progress`
// (reset time passed), the next tick re-enters the normal path and the
// execution is re-driven then.
if (canonicalTask.status === 'rate_limited' || canonicalTask.status === 'usage_limited') {
if (isRateOrUsageLimited(canonicalTask.status)) {
return;
}

Expand Down Expand Up @@ -9986,8 +9987,7 @@ export class SpaceRuntime {
(task) =>
task.status === 'in_progress' ||
task.status === 'approved' ||
task.status === 'rate_limited' ||
task.status === 'usage_limited'
isRateOrUsageLimited(task.status)
).length;
}

Expand Down
9 changes: 4 additions & 5 deletions packages/daemon/src/lib/space/runtime/task-agent-manager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
* registered `onComplete` callbacks are fired.
*/

import { generateUUID, resolveNodeAgents } from '@hyperneo/shared';
import { generateUUID, isRateOrUsageLimited, resolveNodeAgents } from '@hyperneo/shared';
import type {
Space,
SpaceTask,
Expand Down Expand Up @@ -695,7 +695,7 @@ export class TaskAgentManager {
// overwriting those would lose the approval/review lifecycle and the
// post-approval route. The session-level cooldown still holds regardless —
// this guard only protects the task row.
const isLimited = task.status === 'rate_limited' || task.status === 'usage_limited';
const isLimited = isRateOrUsageLimited(task.status);
if (task.status !== 'in_progress' && !isLimited) {
return;
}
Expand Down Expand Up @@ -748,7 +748,7 @@ export class TaskAgentManager {
private async restoreTaskFromRateLimit(taskId: string): Promise<void> {
const task = this.config.taskRepo.getTask(taskId);
if (!task) return;
if (task.status !== 'rate_limited' && task.status !== 'usage_limited') return;
if (!isRateOrUsageLimited(task.status)) return;
this.config.taskRepo.updateTask(taskId, { status: 'in_progress', restrictions: null });
this.emitTaskUpdatedEvent(taskId);
}
Expand Down Expand Up @@ -3641,8 +3641,7 @@ export class TaskAgentManager {
// paused status until the cross-restart sweep restores it — gate on it too.
const parentTaskId = this.findParentTaskIdForSubSession(sessionId);
const parentTask = parentTaskId ? this.config.taskRepo.getTask(parentTaskId) : null;
const parentLimited =
parentTask?.status === 'rate_limited' || parentTask?.status === 'usage_limited';
const parentLimited = parentTask ? isRateOrUsageLimited(parentTask.status) : false;
if ((deliveryMode === 'defer' && isBusy) || inRateLimitCooldown || parentLimited) {
const dbId = this.config.db.saveUserMessage(sessionId, sdkUserMessage, 'deferred', origin);
return dbId;
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import type { NodeExecution, SpaceTask, SpaceWorkflow } from '@hyperneo/shared';
import { resolveNodeAgents } from '@hyperneo/shared';
import { isRateOrUsageLimited, resolveNodeAgents } from '@hyperneo/shared';

export type ExecutionWorkflowValidationResult =
| { valid: true }
Expand Down Expand Up @@ -107,7 +107,7 @@ export function validateTaskAllowsSpawn(task: SpaceTask): void {
// gates the out-of-band activation path (external-event / peer-handoff
// spawns via activateTargetSessionsForMessage), which bypasses the tick loop's
// paused-task guard in processRunTick.
if (task.status === 'rate_limited' || task.status === 'usage_limited') {
if (isRateOrUsageLimited(task.status)) {
// Transient (NOT permanent): the runtime leaves the execution `pending` and
// re-attempts on a later tick once recoverRateLimitedTasks restores the
// task. A PermanentSpawnError here would cancel + unregister the execution,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
*/

import type { Database as BunDatabase } from 'bun:sqlite';
import { generateUUID } from '@hyperneo/shared';
import { generateUUID, isRateOrUsageLimited } from '@hyperneo/shared';
import type {
SpaceTask,
SpaceBlockReason,
Expand Down Expand Up @@ -524,8 +524,7 @@ export class SpaceTaskRepository {
if (
params.restrictions === undefined &&
params.status !== undefined &&
params.status !== 'rate_limited' &&
params.status !== 'usage_limited'
!isRateOrUsageLimited(params.status)
) {
fields.push('restrictions = ?');
values.push(null);
Expand Down
25 changes: 24 additions & 1 deletion packages/shared/src/types/space-utils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,29 @@ export function isWorkflowRunWaiting(status: WorkflowRunStatus | 'failed'): stat
return status === 'blocked';
}

// ============================================================================
// Space task rate/usage-limit paused-status predicate
// ============================================================================

/**
* The task statuses that pause a run on an API rate/usage cap.
*
* A task in one of these statuses still holds its concurrency slot, is not
* spawnable, counts as action-required, and is treated as active by the goal
* runtime — but it normally auto-resumes once the cap lifts.
*
* Single source of truth: every call site that asks "is this a paused-on-cap
* status?" routes through here, so the membership of the paused set is defined
* in one place. Adding or removing such a status is a one-line edit that every
* consumer picks up in lockstep — no duplicated `'rate_limited' ||
* 'usage_limited'` literals to keep in sync across files.
*/
export function isRateOrUsageLimited(
status: SpaceTaskStatus
): status is 'rate_limited' | 'usage_limited' {
return status === 'rate_limited' || status === 'usage_limited';
}

// ============================================================================
// Space task workflow recovery transitions
// ============================================================================
Expand All @@ -109,7 +132,7 @@ export function isWorkflowRecoveryTransition(
(from === 'done' && to === 'in_progress') ||
(from === 'blocked' && (to === 'open' || to === 'in_progress')) ||
(from === 'cancelled' && (to === 'open' || to === 'in_progress')) ||
((from === 'rate_limited' || from === 'usage_limited') && to === 'in_progress')
(isRateOrUsageLimited(from) && to === 'in_progress')
);
}

Expand Down
53 changes: 53 additions & 0 deletions packages/shared/tests/space-utils.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { describe, test, expect } from 'bun:test';
import type {
SpaceTaskStatus,
SpaceWorkerAgent,
SpaceWorkflow,
WorkflowChannel,
Expand All @@ -16,6 +17,7 @@ import {
isWorkflowRunSucceeded,
isWorkflowRunTerminal,
isWorkflowRunWaiting,
isRateOrUsageLimited,
isWorkflowRecoveryTransition,
} from '../src/types/space-utils.ts';

Expand Down Expand Up @@ -380,6 +382,57 @@ describe('getChannelsToNode', () => {
});
});

// ============================================================================
// isRateOrUsageLimited — single source of truth for paused-on-cap statuses
// ============================================================================

describe('isRateOrUsageLimited — single source of truth for paused-on-cap statuses', () => {
const ALL_SPACE_TASK_STATUSES = [
'draft',
'open',
'in_progress',
'review',
'approved',
'done',
'blocked',
'cancelled',
'archived',
'rate_limited',
'usage_limited',
] as const satisfies readonly SpaceTaskStatus[];

// Compile-time exhaustiveness: typecheck error if any SpaceTaskStatus is
// absent from the array above. A newly-added status that isn't listed here
// makes `true` unassignable to `never`, turning a silent coverage hole into a
// build failure — which is the guarantee this test exists to provide.
type _ExhaustiveSpaceTaskStatuses =
SpaceTaskStatus extends (typeof ALL_SPACE_TASK_STATUSES)[number] ? true : never;
const _assertExhaustive: _ExhaustiveSpaceTaskStatuses = true;

test('returns true exactly for rate_limited and usage_limited', () => {
const paused = ALL_SPACE_TASK_STATUSES.filter(isRateOrUsageLimited);
expect([...paused].sort()).toEqual(['rate_limited', 'usage_limited']);
});

test('isWorkflowRecoveryTransition derives its paused arm from the predicate', () => {
// Every consumer of the paused set routes through isRateOrUsageLimited, so
// the recovery transition's rate/usage arm must agree with the predicate
// across the full status set. If a status is added to the predicate, this
// consumer (and every other) picks it up automatically — no per-site
// literal to update. The web consumer (isActionRequired) is covered
// exhaustively in task-filters.test.ts; the daemon consumers by the
// rate-limit integration tests.
for (const status of ALL_SPACE_TASK_STATUSES) {
expect(isWorkflowRecoveryTransition(status, 'in_progress')).toBe(
isRateOrUsageLimited(status) ||
status === 'done' ||
status === 'blocked' ||
status === 'cancelled'
);
}
});
});

// ============================================================================
// isWorkflowRecoveryTransition — paused-task manual resume/cancel routing
// ============================================================================
Expand Down
9 changes: 2 additions & 7 deletions packages/web/src/lib/task-filters.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
* computed signals, and unit tests.
*/

import type { SpaceTask } from '@hyperneo/shared';
import { isRateOrUsageLimited, type SpaceTask } from '@hyperneo/shared';

/**
* Minimum shape needed by `isActionRequired` — a structural subset of
Expand All @@ -31,12 +31,7 @@ export type ActionRequiredTaskInput = Pick<SpaceTask, 'status'>;
* even though it normally auto-resumes when the cap lifts).
*/
export function isActionRequired(task: ActionRequiredTaskInput): boolean {
return (
task.status === 'blocked' ||
task.status === 'review' ||
task.status === 'rate_limited' ||
task.status === 'usage_limited'
);
return task.status === 'blocked' || task.status === 'review' || isRateOrUsageLimited(task.status);
}

/**
Expand Down