mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 00:04:01 +00:00
fix(runtime): reserve restored liveness probe capacity
This commit is contained in:
@@ -10365,7 +10365,7 @@
|
||||
{
|
||||
"file": "src/main/runtime/pty-process-liveness-broker.test.ts",
|
||||
"assertions": [
|
||||
"cross-PTY hanging probes remain below the broker-wide admission cap",
|
||||
"cross-PTY hanging probes remain below the broker-wide admission cap, and a reserved restored-agent lane cannot be starved by unscoped completion/title inspections",
|
||||
"retired restored-agent entries are pruned without deleting evidence shared by other process consumers",
|
||||
"per-identity timeouts, settlement, TTL reuse, backoff, exact absence, and owning-inventory reconciliation remain single-flight"
|
||||
]
|
||||
@@ -10715,7 +10715,7 @@
|
||||
},
|
||||
"performanceBudget": {
|
||||
"required": true,
|
||||
"evidence": "Worker-start reuses the existing one-shot renderer reveal and adds no polling, provider listing, or output work. Startup recovery performs one bounded controller inventory per legacy candidate, exact owner/identity checks, one background reveal, and a provider/renderer snapshot only when an adopted worker's in-memory tail is empty and terminal.read is explicitly requested. One incarnation-scoped process-evidence broker now serves both the incumbent foreground-agent refresh and capable restored-row projection: concurrent callers deduplicate, sequential live reads reuse a 10-second TTL, unavailable reads back off from 3 to 30 seconds, and at most eight unresolved provider probes can exist across PTYs even when their promises never settle. Restored projection builds one O(PTYs + persisted surfaces + restored rows) authority index before inspection and one after it, maps catalog work four-wide, and prunes entries owned only by disappeared restored hooks without pruning shared foreground evidence. Every catalog read shares the existing three-second controller budget. Lifecycle, incarnation, reconnect, and controller changes invalidate evidence. Check adds one optional pane-key field and reuses the existing Run scan or bounded active-Dispatch lookup. Federation replay keeps the existing per-Dispatch polling cadence and existing transport queue limit, retaining a terminal Dispatch only while its imported cursor exceeds one persisted acknowledgment watermark. Each successful ACK adds one primary-key SQLite update; exact eligibility is a primary-key lookup, and identical pending terminal lookup is bounded by the existing 256-item per-Dispatch relay quota and pending-relay index. A version-aware schema sentinel keeps valid v26 databases on the direct v26-to-v27 migration path."
|
||||
"evidence": "Worker-start reuses the existing one-shot renderer reveal and adds no polling, provider listing, or output work. Startup recovery performs one bounded controller inventory per legacy candidate, exact owner/identity checks, one background reveal, and a provider/renderer snapshot only when an adopted worker's in-memory tail is empty and terminal.read is explicitly requested. One incarnation-scoped process-evidence broker now serves both the incumbent foreground-agent refresh and capable restored-row projection: concurrent callers deduplicate, sequential live reads reuse a 10-second TTL, unavailable reads back off from 3 to 30 seconds, at most eight unresolved provider probes can exist across PTYs even when their promises never settle, and at most four are unscoped so restored-agent capacity remains available during hanging title/completion reads. Restored projection builds one O(PTYs + persisted surfaces + restored rows) authority index before inspection and one after it, maps catalog work four-wide, and prunes entries owned only by disappeared restored hooks without pruning shared foreground evidence. Every catalog read shares the existing three-second controller budget. Lifecycle, incarnation, reconnect, and controller changes invalidate evidence. Check adds one optional pane-key field and reuses the existing Run scan or bounded active-Dispatch lookup. Federation replay keeps the existing per-Dispatch polling cadence and existing transport queue limit, retaining a terminal Dispatch only while its imported cursor exceeds one persisted acknowledgment watermark. Each successful ACK adds one primary-key SQLite update; exact eligibility is a primary-key lookup, and identical pending terminal lookup is bounded by the existing 256-item per-Dispatch relay quota and pending-relay index. A version-aware schema sentinel keeps valid v26 databases on the direct v26-to-v27 migration path."
|
||||
},
|
||||
"promotionCriteria": [
|
||||
"Collect 100 consecutive passes or 14 days of stable CI history on macOS, Linux, and Windows.",
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
import type { PtyProcessLivenessBrokerOptions } from './pty-process-liveness-broker'
|
||||
|
||||
export const DEFAULT_LIVE_TTL_MS = 10_000
|
||||
export const DEFAULT_UNAVAILABLE_BACKOFF_BASE_MS = 3_000
|
||||
export const DEFAULT_UNAVAILABLE_BACKOFF_MAX_MS = 30_000
|
||||
export const DEFAULT_MAX_CONCURRENT_PROBES = 8
|
||||
export const DEFAULT_MAX_CONCURRENT_UNSCOPED_PROBES = 4
|
||||
|
||||
export function processEvidenceKey(ptyId: string, identity: string): string {
|
||||
return JSON.stringify([ptyId, identity])
|
||||
}
|
||||
|
||||
export function maxConcurrentProbes(options: PtyProcessLivenessBrokerOptions): number {
|
||||
const configured = options.maxConcurrentProbes ?? DEFAULT_MAX_CONCURRENT_PROBES
|
||||
return Number.isFinite(configured) && configured > 0
|
||||
? Math.max(1, Math.floor(configured))
|
||||
: DEFAULT_MAX_CONCURRENT_PROBES
|
||||
}
|
||||
|
||||
export function maxConcurrentUnscopedProbes(options: PtyProcessLivenessBrokerOptions): number {
|
||||
const configured = options.maxConcurrentUnscopedProbes ?? DEFAULT_MAX_CONCURRENT_UNSCOPED_PROBES
|
||||
return Number.isFinite(configured) && configured > 0
|
||||
? Math.max(1, Math.floor(configured))
|
||||
: DEFAULT_MAX_CONCURRENT_UNSCOPED_PROBES
|
||||
}
|
||||
@@ -122,6 +122,56 @@ describe('PtyProcessLivenessBroker', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('reserves restored-agent capacity from hanging unscoped inspections', async () => {
|
||||
vi.useFakeTimers()
|
||||
const resolvers: ((value: {
|
||||
foregroundProcess: string
|
||||
hasChildProcesses: boolean
|
||||
}) => void)[] = []
|
||||
const inspectProcess = vi.fn(
|
||||
() =>
|
||||
new Promise<{ foregroundProcess: string; hasChildProcesses: boolean }>((resolve) => {
|
||||
resolvers.push(resolve)
|
||||
})
|
||||
)
|
||||
const source = { getForegroundProcess: vi.fn(async () => null), inspectProcess }
|
||||
const broker = new PtyProcessLivenessBroker({ timeoutMs: 100, maxConcurrentProbes: 8 })
|
||||
|
||||
const unscoped = Array.from({ length: 8 }, (_, index) =>
|
||||
broker.inspect({
|
||||
source,
|
||||
ptyId: `unscoped-${index}`,
|
||||
identity: `incarnation-${index}`
|
||||
})
|
||||
)
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
await Promise.all(unscoped)
|
||||
|
||||
expect(inspectProcess).toHaveBeenCalledTimes(4)
|
||||
expect(broker.getActiveUnscopedProbeCount()).toBe(4)
|
||||
const restored = broker.inspect({
|
||||
source,
|
||||
ptyId: 'restored-pty',
|
||||
identity: 'restored-incarnation',
|
||||
consumerId: 'restored-agent'
|
||||
})
|
||||
expect(inspectProcess).toHaveBeenCalledTimes(5)
|
||||
expect(broker.getActiveProbeCount()).toBe(5)
|
||||
resolvers[4]!({ foregroundProcess: 'codex', hasChildProcesses: true })
|
||||
await expect(restored).resolves.toEqual({
|
||||
status: 'live',
|
||||
foregroundProcess: 'codex',
|
||||
hasChildProcesses: true
|
||||
})
|
||||
|
||||
for (const resolve of resolvers) {
|
||||
resolve({ foregroundProcess: 'codex', hasChildProcesses: true })
|
||||
}
|
||||
await vi.runAllTimersAsync()
|
||||
expect(broker.getActiveProbeCount()).toBe(0)
|
||||
expect(broker.getActiveUnscopedProbeCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('caches live evidence across sequential clients', async () => {
|
||||
const inspectProcess = vi.fn(async () => ({
|
||||
foregroundProcess: 'codex',
|
||||
|
||||
@@ -4,6 +4,14 @@ import {
|
||||
type PtyProcessInspectionSource,
|
||||
type PtyProcessLivenessEvidence
|
||||
} from './pty-process-inspection'
|
||||
import {
|
||||
DEFAULT_LIVE_TTL_MS,
|
||||
DEFAULT_UNAVAILABLE_BACKOFF_BASE_MS,
|
||||
DEFAULT_UNAVAILABLE_BACKOFF_MAX_MS,
|
||||
maxConcurrentProbes,
|
||||
maxConcurrentUnscopedProbes,
|
||||
processEvidenceKey
|
||||
} from './pty-process-liveness-broker-state'
|
||||
|
||||
export type {
|
||||
PtyProcessInspectionSource,
|
||||
@@ -13,6 +21,7 @@ export type {
|
||||
type PtyProcessEvidenceEntry = {
|
||||
source: PtyProcessInspectionSource
|
||||
identity: string
|
||||
unscopedProbe: boolean
|
||||
consumerIds: Set<string>
|
||||
hasUnscopedConsumer: boolean
|
||||
freshness: number
|
||||
@@ -27,6 +36,7 @@ type PtyProcessEvidenceEntry = {
|
||||
export type PtyProcessLivenessBrokerOptions = {
|
||||
timeoutMs: number
|
||||
maxConcurrentProbes?: number
|
||||
maxConcurrentUnscopedProbes?: number
|
||||
liveTtlMs?: number
|
||||
unavailableBackoffBaseMs?: number
|
||||
unavailableBackoffMaxMs?: number
|
||||
@@ -34,15 +44,11 @@ export type PtyProcessLivenessBrokerOptions = {
|
||||
onInspectionError?: (ptyId: string, error: unknown) => void
|
||||
}
|
||||
|
||||
const DEFAULT_LIVE_TTL_MS = 10_000
|
||||
const DEFAULT_UNAVAILABLE_BACKOFF_BASE_MS = 3_000
|
||||
const DEFAULT_UNAVAILABLE_BACKOFF_MAX_MS = 30_000
|
||||
const DEFAULT_MAX_CONCURRENT_PROBES = 8
|
||||
|
||||
export class PtyProcessLivenessBroker {
|
||||
private readonly entries = new Map<string, PtyProcessEvidenceEntry>()
|
||||
private readonly now: () => number
|
||||
private activeProbeCount = 0
|
||||
private activeUnscopedProbeCount = 0
|
||||
|
||||
constructor(private readonly options: PtyProcessLivenessBrokerOptions) {
|
||||
this.now = options.now ?? Date.now
|
||||
@@ -123,7 +129,11 @@ export class PtyProcessLivenessBroker {
|
||||
if (waitMs === 0) {
|
||||
return Promise.resolve({ status: 'unverifiable', reason: 'process inspection timed out' })
|
||||
}
|
||||
if (this.activeProbeCount >= this.maxConcurrentProbes()) {
|
||||
const unscopedProbe = args.consumerId === undefined
|
||||
if (
|
||||
this.activeProbeCount >= maxConcurrentProbes(this.options) ||
|
||||
(unscopedProbe && this.activeUnscopedProbeCount >= maxConcurrentUnscopedProbes(this.options))
|
||||
) {
|
||||
return Promise.resolve({
|
||||
status: 'unverifiable',
|
||||
reason: 'process inspection capacity unavailable'
|
||||
@@ -137,6 +147,7 @@ export class PtyProcessLivenessBroker {
|
||||
const entry: PtyProcessEvidenceEntry = {
|
||||
source: args.source,
|
||||
identity: args.identity,
|
||||
unscopedProbe,
|
||||
consumerIds: new Set(args.consumerId ? [args.consumerId] : []),
|
||||
hasUnscopedConsumer: args.consumerId === undefined,
|
||||
freshness,
|
||||
@@ -148,6 +159,9 @@ export class PtyProcessLivenessBroker {
|
||||
probe: null
|
||||
}
|
||||
this.activeProbeCount += 1
|
||||
if (unscopedProbe) {
|
||||
this.activeUnscopedProbeCount += 1
|
||||
}
|
||||
const probe = inspectPtyProcess(args.source, args.ptyId)
|
||||
.then((evidence) => {
|
||||
const reconciled =
|
||||
@@ -195,6 +209,9 @@ export class PtyProcessLivenessBroker {
|
||||
})
|
||||
entry.probe = probe.finally(() => {
|
||||
this.activeProbeCount -= 1
|
||||
if (entry.unscopedProbe) {
|
||||
this.activeUnscopedProbeCount -= 1
|
||||
}
|
||||
})
|
||||
this.entries.set(args.ptyId, entry)
|
||||
return this.waitForProbe(args.ptyId, entry, waitMs)
|
||||
@@ -243,6 +260,10 @@ export class PtyProcessLivenessBroker {
|
||||
return this.activeProbeCount
|
||||
}
|
||||
|
||||
getActiveUnscopedProbeCount(): number {
|
||||
return this.activeUnscopedProbeCount
|
||||
}
|
||||
|
||||
getEntryCount(): number {
|
||||
return this.entries.size
|
||||
}
|
||||
@@ -281,13 +302,6 @@ export class PtyProcessLivenessBroker {
|
||||
return Math.min(max, base * 2 ** Math.max(0, failureCount - 1))
|
||||
}
|
||||
|
||||
private maxConcurrentProbes(): number {
|
||||
const configured = this.options.maxConcurrentProbes ?? DEFAULT_MAX_CONCURRENT_PROBES
|
||||
return Number.isFinite(configured) && configured > 0
|
||||
? Math.max(1, Math.floor(configured))
|
||||
: DEFAULT_MAX_CONCURRENT_PROBES
|
||||
}
|
||||
|
||||
private rememberConsumer(entry: PtyProcessEvidenceEntry, consumerId: string | undefined): void {
|
||||
if (consumerId === undefined) {
|
||||
entry.hasUnscopedConsumer = true
|
||||
@@ -302,7 +316,3 @@ export class PtyProcessLivenessBroker {
|
||||
entry.expiresAt = this.now() + this.unavailableBackoffMs(entry.failureCount)
|
||||
}
|
||||
}
|
||||
|
||||
function processEvidenceKey(ptyId: string, identity: string): string {
|
||||
return JSON.stringify([ptyId, identity])
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user